ableegoldman commented on a change in pull request #9157:
URL: https://github.com/apache/kafka/pull/9157#discussion_r478730765
##########
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 think with this replacement then we might be able to get out of doing
any kind of special handling for the combined window outside of `processEarly`
----------------------------------------------------------------
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:
us...@infra.apache.org
