ableegoldman commented on a change in pull request #9157:
URL: https://github.com/apache/kafka/pull/9157#discussion_r478762667
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamSlidingWindowAggregate.java
##########
@@ -174,12 +181,11 @@ public void processInOrder(final K key, final V value,
final long timestamp) {
}
}
}
-
//create right window for previous record
if (latestLeftTypeWindow != null) {
- final long leftWindowEnd =
latestLeftTypeWindow.key.window().end();
- final long rightWinStart = leftWindowEnd ==
windows.timeDifferenceMs() ? latestLeftTypeWindow.value.timestamp() + 1 :
leftWindowEnd + 1;
- if (!windowStartTimes.contains(rightWinStart)) {
+ final long previousRecord =
latestLeftTypeWindow.key.window().end();
+ final long rightWinStart = previousRecord ==
windows.timeDifferenceMs() ? latestLeftTypeWindow.value.timestamp() + 1 :
previousRecord + 1;
Review comment:
Ok before I continue to try and wrap my head around this, is this
particular line going to be moot if we start tracking `previousRecordTimestamp`
instead of `latestLeftTypeWindow`? The point of it was just to distinguish that
special case, but now we can just say `rightWindowStart =
previousRecordTimestamp + 1` -- does that sound right?
----------------------------------------------------------------
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
