lct45 commented on a change in pull request #9157:
URL: https://github.com/apache/kafka/pull/9157#discussion_r478435832
##########
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()) {
latestLeftTypeWindow = next;
}
} else if (endTime == timestamp) {
leftWinAlreadyCreated = true;
+ // if current record's left window is the combined
window, need to check later if there is a
+ // record that needs a right window within the
combined window
+ if (endTime == windows.timeDifferenceMs()) {
+ latestLeftTypeWindow = next;
+ }
Review comment:
I see what you're saying about checking to see if the left window (in
this case [0,timeDifference]) was already a left window and I can substitute
for that check. The overall point was to catch instances of a record coming in
at say 10 (with a timeDifference of 10) and also having one at 5, so 5 needs a
right window but the combined window will only be picked up by this `if()`
since `endTime == timestamp`. Setting `latestLeftTypeWindow = next` here will
generally _not_ be a left type window, but the combined window is a special
case since there are records that may need a right window in there anyways.
Because this tracks the _latest_ left type window, storing the combined window
should never override any other formal left type window that needs a right
window.
That being said, I get that this is confusing. Do you think changing the
check to `if (endTime == windows.TimeDifferenceMs() && !isLeftWindow(next))
would make it seem cleaner?
----------------------------------------------------------------
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]
