izzyacademy commented on a change in pull request #10740:
URL: https://github.com/apache/kafka/pull/10740#discussion_r655661049
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
##########
@@ -84,17 +86,14 @@ private SessionWindows(final long gapMs, final long
graceMs) {
* Create a new window specification with the specified inactivity gap.
*
* @param inactivityGap the gap of inactivity between sessions
- * @return a new window specification with default maintain duration of 1
day
+ * @return a new window specification with default without any grace period
Review comment:
@ableegoldman I will change it now to DEPRECATED_OLD_24_HR_GRACE_PERIOD
and for the zero grace period, I will use NO_GRACE_PERIOD as the constant
##########
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.
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java
##########
@@ -36,11 +36,11 @@
* @see SessionWindows
* @see TimestampExtractor
*/
+@SuppressWarnings("deprecation")
public abstract class Windows<W extends Window> {
- // By default grace period is 24 hours for all windows,
- // in other words we allow out-of-order data for up to a day
- protected static final long DEFAULT_GRACE_PERIOD_MS = 24 * 60 * 60 * 1000L;
+ // By default, there is no longer any grace period since Kafka 3.0
+ protected static final long DEFAULT_GRACE_PERIOD_MS = 0L;
Review comment:
I am going to address that now, with the new code version
--
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:
[email protected]
