KStreamWindowAggregate 2.0.1 vs 2.5.0: skipping records instead of processing

user_6483435 0
Viewed 361

I've recently upgraded my kafka streams from 2.0.1 to 2.5.0. As a result I'm seeing a lot of warnings like the following:

org.apache.kafka.streams.kstream.internals.KStreamWindowAggregate$KStreamWindowAggregateProcessor Skipping record for expired window. key=[325233] topic=[MY_TOPIC] partition=[20] offset=[661798621] timestamp=[1600041596350] window=[1600041570000,1600041600000) expiration=[1600059629913] streamTime=[1600145999913]

There seem to be new logic in the KStreamWindowAggregate class that checks if a window has closed. If it has been closed the messages are skipped. Compared to 2.0.1 these messages where still processed.

Question

Is there a way to get the same behavior like before? I'm seeing lots of gaps in my data with this upgrade and not sure how to solve this, as previously these gaps where not seen.

The aggregate function that I'm using already deals with windowing and as a result with expired windows. How does this new logic relate to this expiring windows?

Update

While further exploring I indeed see it to be related to the graceperiod in ms. It seems that in my custom timestampextractor (that has the logic to use the timestamp from the payload instead of the normal timestamp), I'm able to see that the incoming timestamp for the expired window warnings indeed is bigger than the 24 hours compared to the event time from the payload.

I assume this is caused by consumer lags of over 24 hours.

The timestamp extractor extract method has a partition time which according to the docs:

partitionTime the highest extracted valid timestamp of the current record's partition˙ (could be -1 if unknown)

so is this the create time of the record on the topic? And is there a way to influence this in a way that my records are no longer skipped?

apache-kafka-streams apache-kafka
1 Answers

Compared to 2.0.1 these messages where still processed.

That is a little bit surprising (even if I would need to double check the code), at least for the default config. By default, store retention time is set to 24h, and thus in 2.0.1 older messages than 24h should also not be processed as the corresponding state got purged already. If you did change the store retention time (via Materialized#withRetention) to a larger value, you would also need to increase the window grace period via TimeWindows#grace() method accordingly.

The aggregate function that I'm using already deals with windowing and as a result with expired windows. How does this new logic relate to this expiring windows?

Not sure what you mean by this or how you actually do this? The old and new logic are similar with regard to how a long a window is stored (retention time config). The new part is the grace period that you can increase to the same value as retention time if you wish).

About "partition time": it is computed base on whatever TimestampExtractor returns. For your case, it's the max of whatever you extracted from the message payload.

user_49530790
Related