spena commented on a change in pull request #10462:
URL: https://github.com/apache/kafka/pull/10462#discussion_r610044912
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java
##########
@@ -38,20 +48,36 @@
private final String otherWindowName;
private final long joinBeforeMs;
private final long joinAfterMs;
+ private final long joinGraceMs;
private final ValueJoinerWithKey<? super K, ? super V1, ? super V2, ?
extends R> joiner;
private final boolean outer;
+ private final Optional<String> outerJoinWindowName;
+ private final boolean thisJoin;
+
+ // Observed time is AtomicLong because this time is shared between the
left and side processor nodes. However,
+ // this time is not updated in parallel, so we can call get() several
times without worry about getting different
+ // times.
+ private final AtomicLong maxObservedStreamTime;
Review comment:
I think `context.currentStreamTimeMs()` should work. I wasn't awrare I
could get the new stream time from it. I don't see any problems as I only need
the stream time to expire the records.
--
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
