raminqaf commented on code in PR #15601:
URL: https://github.com/apache/kafka/pull/15601#discussion_r1593500243
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -221,41 +199,28 @@ private void emitNonJoinedOuterRecords(
// reset to MAX_VALUE in case the store is empty
sharedTimeTracker.minTime = Long.MAX_VALUE;
- try (final KeyValueIterator<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<V1, V2>> it = store.all()) {
+ try (final KeyValueIterator<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<VL, VR>> it = store.all()) {
TimestampedKeyAndJoinSide<K> prevKey = null;
- boolean outerJoinLeftWindowOpen = false;
- boolean outerJoinRightWindowOpen = false;
while (it.hasNext()) {
- if (outerJoinLeftWindowOpen && outerJoinRightWindowOpen) {
- // if windows are open for both joinSides we can break
since there are no more candidates to emit
+ final KeyValue<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<VL, VR>> nextKeyValue = it.next();
+ final TimestampedKeyAndJoinSide<K>
timestampedKeyAndJoinSide = nextKeyValue.key;
+ sharedTimeTracker.minTime =
timestampedKeyAndJoinSide.getTimestamp();
+ if
(isOuterJoinWindowOpenForSide(timestampedKeyAndJoinSide, true) &&
isOuterJoinWindowOpenForSide(timestampedKeyAndJoinSide, false)) {
Review Comment:
Yeah, you are right! That's a good catch. I was depending too much on the
tests. I couldn't find any test that triggered the condition even after
reverting the changes.
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
