lct45 commented on a change in pull request #9157:
URL: https://github.com/apache/kafka/pull/9157#discussion_r478445137
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java
##########
@@ -160,11 +160,18 @@ public void processInOrder(final K key, final V value,
final long timestamp) {
if (endTime < timestamp) {
leftWinAgg = next.value;
+ // store the combined window if it is found so that a
right window can be created for
+ // the combined window's max record, as needed
if (isLeftWindow(next) || endTime ==
windows.timeDifferenceMs()) {
Review comment:
It is, until we have a combined window that holds records that don't
have corresponding left windows. I treated the combined window as a defacto
left window since it's taking the place of early record's left windows. EX: for
`timeDifference=10`, a record at 4 would be stored in [0,10]. If a record comes
in at 11, we need the window from [5,15], but we only create that if we find a
`leftTypeWindow`, which doesn't exist for 4. For these records, [0,10] is the
only window we have, and therefore the closest thing to a left type window.
I didn't change variable names because I wanted the original algorithm to be
understandable. It feels like adding early records has added a significant
amount of complexity and I do wonder if it's worth it to have the addition
confusion for testing flexibility - WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
