[jira] [Created] (KAFKA-13739) Sliding window without grace not working

["bounkong khamphousone (Jira)" <ji...@apache.org>]

bounkong khamphousone created KAFKA-13739:
---------------------------------------------

             Summary: Sliding window without grace not working
                 Key: KAFKA-13739
                 URL: https://issues.apache.org/jira/browse/KAFKA-13739
             Project: Kafka
          Issue Type: Bug
          Components: streams
    Affects Versions: 3.1.0
            Reporter: bounkong khamphousone


Hi everyone! I would like to understand why KafkaStreams DSL offer the ability to express a SlidingWindow with no grace period but seems that it doesn't work. [confluent's site|https://docs.confluent.io/platform/current/streams/developer-guide/dsl-api.html#sliding-time-windows] state that grace period is required and with the deprecated method, it's default to 24 hours.

Doing a basic sliding window with a count, if I set grace period to 1 ms, expected output is done. Based on the sliding window documentation, lower and upper bounds are inclusive.

If I set grace period to 0 ms, I can see that record is not skipped at KStreamSlidingWindowAggregate(l.126) but when we try to create the window and push the event in KStreamSlidingWindowAggregate#createWindows we call the method updateWindowAndForward(l.417). This method (l.468) check that {{{}windowEnd > closeTime{}}}.

closeTime is defined as {{observedStreamTime - window.gracePeriodMs}} (Sliding window configuration)

windowEnd is defined as {{{}inputRecordTimestamp{}}}.

 

For a first event with a record timestamp, we can assume that observedStreamTime is equal to inputRecordTimestamp.

 

Therefore, closeTime is {{inputRecordTimestamp - 0}} (gracePeriodMS) which results to {{{}inputRecordTimestamp{}}}.

If we go back to the check done in {{updateWindowAndForward}} method, then we have inputRecordTimestamp > inputRecordTimestamp which is always false. The record is then skipped for record's own window.

Stating that lower and upper bounds are inclusive, I would have expected the event to be pushed in the store and forwarded. Hence, the check would be {{{}windowEnd >= closeTime{}}}.

 

Is it a bug or is it intended ?

Thanks in advance for your explanations!

Best regards!



--
This message was sent by Atlassian Jira
(v8.20.1#820001)