izzyacademy commented on a change in pull request #10740:
URL: https://github.com/apache/kafka/pull/10740#discussion_r655661436
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
##########
@@ -119,6 +120,50 @@ public SessionWindows grace(final Duration afterWindowEnd)
throws IllegalArgumen
return new SessionWindows(gapMs, afterWindowEndMs);
}
+ /**
+ * Create a new window specification with the specified inactivity gap.
+ *
+ * @param inactivityGap the gap of inactivity between sessions
+ * @return a new window specification without any grace period
+ *
+ * @throws IllegalArgumentException if {@code inactivityGap} is zero or
negative or can't be represented as {@code long milliseconds}
+ * @since 3.0
+ */
+ public static SessionWindows ofInactivityGapWithNoGrace(final Duration
inactivityGap) {
+ return ofInactivityGapAndGrace(inactivityGap,
ofMillis(DEFAULT_GRACE_PERIOD_MS));
+ }
+
+ /**
+ * Reject out-of-order events that arrive more than {@code afterWindowEnd}
+ * after the end of its window.
+ * <p>
+ * Note that new events may change the boundaries of session windows, so
aggressive
+ * close times can lead to surprising results in which an out-of-order
event is rejected and then
+ * a subsequent event moves the window boundary forward.
+ *
+ * @param inactivityGap the gap of inactivity between sessions
+ * @param afterWindowEnd The grace period to admit out-of-order events to
a window.
+ * @return A SessionWindows object with the specified inactivity gap and
grace period
+ * @throws IllegalArgumentException if the {@code afterWindowEnd} is
negative of can't be represented as {@code long milliseconds}
+ * @since 3.0
+ */
+ public static SessionWindows ofInactivityGapAndGrace(final Duration
inactivityGap, final Duration afterWindowEnd) {
+
+ final String msgPrefixInactivityGapMs =
prepareMillisCheckFailMsgPrefix(inactivityGap, "inactivityGap");
+ final long inactivityGapMs =
validateMillisecondDuration(inactivityGap, msgPrefixInactivityGapMs);
+ if (inactivityGapMs <= 0) {
+ throw new IllegalArgumentException("Gap time (inactivityGapMs)
cannot be zero or negative.");
+ }
+
+ final String msgPrefixAfterWindowEnd =
prepareMillisCheckFailMsgPrefix(afterWindowEnd, "afterWindowEnd");
+ final long afterWindowEndMs =
validateMillisecondDuration(afterWindowEnd, msgPrefixAfterWindowEnd);
+ if (afterWindowEndMs < 0) {
+ throw new IllegalArgumentException("Grace period must not be
negative.");
+ }
Review comment:
This is a great idea. Makes sense.
--
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
