VictorvandenHoven commented on code in PR #14426: URL: https://github.com/apache/kafka/pull/14426#discussion_r1474063977
########## streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamLeftJoinTest.java: ########## @@ -609,17 +842,18 @@ public void testOrdering() { inputTopic1.pipeInput(1, "A1", 100L); processor.checkAndClearProcessResult(); - // push one item to the other window that has a join; this should produce non-joined records with a closed window first, then - // the joined records + // push one item to the other window that has a join; + // this should produce the joined record first; + // then non-joined record with a closed window // by the time they were produced before // w1 = { 0:A0 (ts: 0), 1:A1 (ts: 100) } // w2 = { } // --> w1 = { 0:A0 (ts: 0), 1:A1 (ts: 100) } // --> w2 = { 1:a1 (ts: 110) } inputTopic2.pipeInput(1, "a1", 110L); processor.checkAndClearProcessResult( - new KeyValueTimestamp<>(0, "A0+null", 0L), - new KeyValueTimestamp<>(1, "A1+a1", 110L) + new KeyValueTimestamp<>(1, "A1+a1", 110L), + new KeyValueTimestamp<>(0, "A0+null", 0L) Review Comment: What we have seen is that you could emit both a null-joined record and a joined-record if you first check the outerjoinstore (via emitNonJoinedOuterRecords()) and only after that, a record from the otherWindowStore. This behaviour is not what you want. Therefore you should first find out if there is going to be a joined record and if there is, you have to nullify this in the relevant outerjoinstore. -- 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