guozhangwang commented on a change in pull request #10462:
URL: https://github.com/apache/kafka/pull/10462#discussion_r608905403
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java
##########
@@ -133,34 +141,37 @@ public void process(final K key, final V1 value) {
return;
}
- // maxObservedStreamTime is updated and shared between left and
right sides, so we can
- // process a non-join record immediately if it is late
- final long maxStreamTime = maxObservedStreamTime.updateAndGet(time
-> Math.max(time, context().timestamp()));
-
boolean needOuterJoin = outer;
+ boolean joinFound = false;
final long inputRecordTimestamp = context().timestamp();
final long timeFrom = Math.max(0L, inputRecordTimestamp -
joinBeforeMs);
final long timeTo = Math.max(0L, inputRecordTimestamp +
joinAfterMs);
+ // maxObservedStreamTime is updated and shared between left and
right join sides
+ maxObservedStreamTime.updateAndGet(t -> Math.max(t,
inputRecordTimestamp));
+
try (final WindowStoreIterator<V2> iter = otherWindow.fetch(key,
timeFrom, timeTo)) {
while (iter.hasNext()) {
needOuterJoin = false;
+ joinFound = true;
final KeyValue<Long, V2> otherRecord = iter.next();
final long otherRecordTimestamp = otherRecord.key;
+
+ // Emit expired records before the joined record to keep
time ordering
Review comment:
cc @mjsax as well, LMK WDYT.
--
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
