Mailing-List: contact jira-help@kafka.apache.org; run by ezmlm
Precedence: bulk
Reply-To: jira@kafka.apache.org
Date: Sat, 26 Aug 2017 17:57:00 +0000 (UTC)
From: "Richard Yu (JIRA)" <jira@apache.org>
To: jira@kafka.apache.org
Message-ID: <JIRA.13024262.1480464424000.129497.1503770220163@Atlassian.JIRA>
In-Reply-To: <JIRA.13024262.1480464424000@Atlassian.JIRA>
References: <JIRA.13024262.1480464424000@Atlassian.JIRA> <JIRA.13024262.1480464424641@jira-lw-us.apache.org>
Subject: [jira] [Comment Edited] (KAFKA-4468) Correctly calculate the window
 end timestamp after read from state stores
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: 7bit
archived-at: Sat, 26 Aug 2017 17:57:09 -0000


    [ https://issues.apache.org/jira/browse/KAFKA-4468?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16142877#comment-16142877 ] 

Richard Yu edited comment on KAFKA-4468 at 8/26/17 5:56 PM:
------------------------------------------------------------

When looking through the references for Windowed Deserializer, it appears that there was none other than unit tests. 

The other approach is to look through the Window Store classes and look for the Window size. 
Inherently, since RocksDBWindowStore is the class that is most commonly used. I looked through that particular class
and found that when instantiating RocksDBWindowStore's window. They called the class WindowStoreIteratorWrapper. 
In lines 54 - 58 of that particular class, this is what I found:

{code}
@Override
public Windowed<Bytes> peekNextKey() {
        final Bytes next = bytesIterator.peekNextKey();
        final long timestamp = WindowStoreUtils.timestampFromBinaryKey(next.get());
        final Bytes key = WindowStoreUtils.bytesKeyFromBinaryKey(next.get());
        return new Windowed<>(key, WindowStoreUtils.timeWindowForSize(timestamp, windowSize));
}
{code}

In this method, {code} windowSize {code} is known to the class already.  Therefore, this is a successful implementation of
using a window of limited time alive, since the fixed size is already known. 

In other words, for us to find the fixed length of time that is needed to be alive, it must be found in the series of calls in which WindowedDeserializer is called. However, when attempting to look for find this sequence of calls between Window Store and WindowedDeserializer, it could not be found.

In other words, their might not be a practical way to retrieve the length of the window.


was (Author: yohan123):
When looking through the references for Windowed Deserializer, it appears that there was none other than unit tests. 

The other approach is to look through the Window Store classes and look for the Window size. 
Inherently, since RocksDBWindowStore is the class that is most commonly used. I looked through that particular class
and found that when instantiating RocksDBWindowStore's window. They called the class WindowStoreIteratorWrapper. 
In lines 54 - 58 of that particular class, this is what I found:

{code}
@Override
public Windowed<Bytes> peekNextKey() {
        final Bytes next = bytesIterator.peekNextKey();
        final long timestamp = WindowStoreUtils.timestampFromBinaryKey(next.get());
        final Bytes key = WindowStoreUtils.bytesKeyFromBinaryKey(next.get());
        return new Windowed<>(key, WindowStoreUtils.timeWindowForSize(timestamp, windowSize));
}
{code}

In this method, {code} windowSize {code} is known to the class already.  Therefore, this is a successful implementation of
using a window of limited time alive, since the fixed size is already known. 

In other words, for us to find the fixed length of time that is needed to be alive, it must be found in the series of calls in which WindowedDeserializer is called. However, when attempting to look for find this sequence of calls between Window Store and WindowedDeserializer. Yet, to date, I could not find such a series of calls.

In other words, their might not be a practical way to retrieve the length of the window.

> Correctly calculate the window end timestamp after read from state stores
> -------------------------------------------------------------------------
>
>                 Key: KAFKA-4468
>                 URL: https://issues.apache.org/jira/browse/KAFKA-4468
>             Project: Kafka
>          Issue Type: Bug
>          Components: streams
>            Reporter: Guozhang Wang
>              Labels: architecture
>
> When storing the WindowedStore on the persistent KV store, we only use the start timestamp of the window as part of the combo-key as (start-timestamp, key). The reason that we do not add the end-timestamp as well is that we can always calculate it from the start timestamp + window_length, and hence we can save 8 bytes per key on the persistent KV store.
> However, after read it (via {{WindowedDeserializer}}) we do not set its end timestamp correctly but just read it as an {{UnlimitedWindow}}. We should fix this by calculating its end timestamp as mentioned above.


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)