VictorvandenHoven commented on code in PR #15510:
URL: https://github.com/apache/kafka/pull/15510#discussion_r1521827156
##########
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java:
##########
@@ -223,9 +223,9 @@ private void emitNonJoinedOuterRecords(
try (final KeyValueIterator<TimestampedKeyAndJoinSide<K>,
LeftOrRightValue<V1, V2>> it = store.all()) {
TimestampedKeyAndJoinSide<K> prevKey = null;
+ boolean outerJoinLeftBreak = false;
+ boolean outerJoinRightBreak = false;
while (it.hasNext()) {
- boolean outerJoinLeftBreak = false;
- boolean outerJoinRightBreak = false;
Review Comment:
Currently, in the while-loop of the emitNonJoinedOuterRecords() method we
are iterating over ALL the left- and right-side outerJoinRecords that are
available in the outerjoin-store until we meet the **break;**.
The idea of the two outerJoinBreak flags was to keep track of when the
window of the outerJoinRecords is not closed anymore, but this is only useful
if the ordering of the KeyValueIterator is by timeStampedKey, and it is not:
The documentation says that the ordering of the KeyValueIterator is NOT
guaranteed.
So, now I think we better can remove the outerJoinBreak flags and just check
for each outerJoinRecord whether it belongs to a closed window or not, without
any optimization.
If the window has closed we can emit a nullJoinedValue. If the window is not
closed yet we can continue with the next outerJoinRecord.
What do you think? @mjsax @florin-akermann
--
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
