mjsax commented on code in PR #14426:
URL: https://github.com/apache/kafka/pull/14426#discussion_r1491788610
##########
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:
I agree that we should not produce a null-joined record and joined-record.
Can you elaborate on the case when (and why) this could exactly happen? It
sounds like a bug in `emitNonJoinedOuterRecords()` to me.
When we check the `outerStore` we should only emit null-joined-records that
are "expired", ie, which belong to already closed windows, and thus, the order
in which we check should not matter. It sounds like as if we might incorrectly
emit null-joined records of window that are not closed yet (would be a bug) or
for windows which are already closed but we emit an incorrect join-record.
Given that we identified that we indeed have a bug with regard to "late
records" and that we don't respect the grace-period correctly, I would assume
it's the same root cause.
Thus, I am now wondering if we should merge this PR as-is (and fix the order
back in a follow PR @florin-akermann is doing), or revert this change in this
PR right away, or first try to merge Florin's PR and re-evaluate this change
afterwards?
Thoughts?
--
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
