guozhangwang commented on a change in pull request #11114:
URL: https://github.com/apache/kafka/pull/11114#discussion_r675318036
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
##########
@@ -102,49 +102,61 @@ private JoinWindows(final long beforeMs,
/**
* Specifies that records of the same key are joinable if their timestamps
are within {@code timeDifference},
- * i.e., the timestamp of a record from the secondary stream is max {@code
timeDifference} earlier or later than
- * the timestamp of the record from the primary stream. Using the method
explicitly sets the grace period to
- * the duration specified by {@code afterWindowEnd} which means that out
of order records arriving
- * after the window end will be dropped. The delay is defined as
(stream_time - record_timestamp).
+ * i.e., the timestamp of a record from the secondary stream is max {@code
timeDifference} before or after
+ * the timestamp of the record from the primary stream.
+ * <p>
+ * Using this method explicitly sets the grace period to the duration
specified by {@code afterWindowEnd}, which
+ * means that only out-of-order records arriving more than the grace
period after the window end will be dropped.
+ * The window close, after which any incoming records are considered late
and will be rejected, is defined as
+ * {@code windowEnd + afterWindowEnd}
*
* @param timeDifference join window interval
* @param afterWindowEnd The grace period to admit out-of-order events to
a window.
- * @throws IllegalArgumentException if the {@code afterWindowEnd} is
negative of can't be represented as {@code long milliseconds}
* @return A new JoinWindows object with the specified window definition
and grace period
+ * @throws IllegalArgumentException if {@code timeDifference} is negative
or can't be represented as {@code long milliseconds}
+ * if the {@code afterWindowEnd} is
negative or can't be represented as {@code long milliseconds}
*/
public static JoinWindows ofTimeDifferenceAndGrace(final Duration
timeDifference, final Duration afterWindowEnd) {
- return new JoinWindows(timeDifference.toMillis(),
timeDifference.toMillis(), afterWindowEnd.toMillis(), true);
+ final String timeDifferenceMsgPrefix =
prepareMillisCheckFailMsgPrefix(timeDifference, "timeDifference");
Review comment:
Thanks for the catch!
--
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
