vvcephei commented on a change in pull request #10378:
URL: https://github.com/apache/kafka/pull/10378#discussion_r602602147
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
##########
@@ -73,33 +72,16 @@
public final class SessionWindows {
private final long gapMs;
- private final long maintainDurationMs;
- private final long graceMs;
+ // By default grace period is 24 hours,
+ // in other words we allow out-of-order data for up to a day
Review comment:
Not to be nitpicky, but should we have comments like this in places like
this? If we change the default later, it might be hard to track down all the
comments that need to be updated.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
##########
@@ -70,57 +69,24 @@
*/
public final class JoinWindows extends Windows<Window> {
- private final long maintainDurationMs;
-
/** Maximum time difference for tuples that are before the join tuple. */
public final long beforeMs;
/** Maximum time difference for tuples that are after the join tuple. */
public final long afterMs;
+ // By default grace period is 24 hours,
Review comment:
I think that's extremely reasonable.
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
##########
@@ -224,55 +161,20 @@ public long size() {
* @return this updated builder
* @throws IllegalArgumentException if the {@code afterWindowEnd} is
negative of can't be represented as {@code long milliseconds}
*/
- @SuppressWarnings("deprecation") // removing segments from Windows will
fix this
public JoinWindows grace(final Duration afterWindowEnd) throws
IllegalArgumentException {
final String msgPrefix =
prepareMillisCheckFailMsgPrefix(afterWindowEnd, "afterWindowEnd");
final long afterWindowEndMs =
validateMillisecondDuration(afterWindowEnd, msgPrefix);
if (afterWindowEndMs < 0) {
throw new IllegalArgumentException("Grace period must not be
negative.");
}
- return new JoinWindows(beforeMs, afterMs, afterWindowEndMs,
maintainDurationMs, segments);
+ return new JoinWindows(beforeMs, afterMs, afterWindowEndMs);
}
@Override
public long gracePeriodMs() {
- // NOTE: in the future, when we remove maintainMs,
- // we should default the grace period to 24h to maintain the default
behavior,
- // or we can default to (24h - size) if you want to be super accurate.
Review comment:
It feels so good to see this go...
##########
File path: streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java
##########
@@ -42,66 +38,10 @@
*/
public abstract class Windows<W extends Window> {
- private long maintainDurationMs = DEFAULT_RETENTION_MS;
- @Deprecated public int segments = 3;
+ protected static final long DEFAULT_GRACE_PERIOD_MS = 24 * 60 * 60 *
1000L; // one day
Review comment:
Great!
Do you think we can write down that this class should effectively be treated
as an interface (no instance members), but that it's kept as an abstract class
purely for compatibility with old versions?
Even better than a comment would be to add a unit test that will fail if
someone declares a non-static member in this class. Looking at the reflection
API, it looks like you can take the union of: `getDeclaredFields() ++
getFields() ++ getDeclaredMethods() ++ getMethods()` (cf
https://docs.oracle.com/javase/tutorial/reflect/class/classMembers.html)
This will give you a set of `Member`s (`Methods` and `Fields`). You could
then use `Member#getModifiers()`
(https://docs.oracle.com/javase/8/docs/api/java/lang/reflect/Member.html#getModifiers--)
and verify that `Modifier.isStatic(modifier)`
(https://docs.oracle.com/javase/8/docs/api/java/lang/reflect/Modifier.html#isStatic-int-)
is `true` for every member.
Not sure if there's a more direct way. Also, it's up to you whether you go
this way or not... I'd just be a little sad if we get right back into this mess
immediately after finally untangling it just because some reviewer didn't
remember the history here.
##########
File path:
streams/src/test/java/org/apache/kafka/streams/kstream/WindowsTest.java
##########
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.kafka.streams.kstream;
Review comment:
Oooh! Or maybe we could use it to enforce that Windows is really a pure
abstract class (my idea above). :)
--
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
