kafka-streams sliding agg window discard out-of-order record belongs to window without grace

user3706408 0
Viewed 53

I have next error, actually i don't understand it.

o.a.k.s.k.i.KStreamSlidingWindowAggregate - Skipping record for expired window. topic=[...] partition=[0] offset=[16880] timestamp=[1662556875000] window=[1662542475000,1662556875000] expiration=[1662556942000] streamTime=[1662556942000]

streamTime=[1662556942000]

timestamp=[1662556875000]

streamTime-timestamp = 67s

window size is 4hour.

grace period is 0

Why was record skipped and i didn't get a output message? it belongs to window. Yes record out-of-order

Update: After read more about kafka-streams, i understand that on each message it creates two window:

  1. (message time - window) and this window include message.
  2. (message time + window) and this window exclude message.

Window 1 is expired. Window 2 don't. that's why i dind't see out message.

But logically it's wrong, message belong to window but i havn't a out message.

Example

sliding window time diff = 10, grace = 0


stream time = 0
send message (time = 10, key = 2) -> message key = 2; stream time = 10
send message (time = 4, key = 1) -> no out message; 
send message (time = 5, key = 1) -> no out message; 

last message belongs to window (stream-time - window-time)

------ restart stream -------

stream time = 0
send message (time = 10, key = 2) -> message key = 2; stream time = 10
send message (time = 4, key = 2) -> 2 message
apache-kafka-streams apache-kafka
1 Answers

In Kafka Streams sliding windows are event based. A new window is created each time a new record enters or drops from the window. It is defined by the record timestamp and a fixed duration.
Each record creates a window [record.timestamp - duration, record.timestamp] and
Each dropped record creates a window [record.timestamp + 1ms, record.timestamp + 1ms + duration].
(Be aware that other stream processing frameworks use a totally different definition of 'sliding windows')

The record is not included in the window when stream-time > window-end + grace-period (https://kafka.apache.org/27/javadoc/org/apache/kafka/streams/kstream/SlidingWindows.html)

For your initial example, the grace-period is zero and your window ends(at record timestamp) after the stream-time; thus the record is not included in the window.

For the second example, I am not sure. My guess is the records with key=1 are expired because the stream-time(10) has exceeded the record times(4,5)(and grace period=0). For the records with key=2, one window is created for the record with timestamp=10 and an update of the same window is emitted, because the record satisfies the condition above. However, no additional windows are created for the out-of-order record, because the grace period is zero.

WnaJ0
Related