You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2021/03/23 04:50:58 UTC
[GitHub] [kafka] guozhangwang opened a new pull request #10378: KAFKA-7106: remove deprecated Windows APIs
guozhangwang opened a new pull request #10378:
URL: https://github.com/apache/kafka/pull/10378
1) Remove all deprecated APIs in KIP-328.
2) Remove deprecated APIS in Windows in KIP-358.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #10378: KAFKA-7106: remove deprecated Windows APIs
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #10378:
URL: https://github.com/apache/kafka/pull/10378#discussion_r602811233
##########
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:
That's actually a great point. I will remove all these on inherited classes and only leave a comment on top of `DEFAULT_GRACE_PERIOD_MS`. When we change it later, we only need to change the single comment there.
--
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:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on pull request #10378: KAFKA-7106: remove deprecated Windows APIs
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on pull request #10378:
URL: https://github.com/apache/kafka/pull/10378#issuecomment-808937641
Bumped into https://issues.apache.org/jira/browse/KAFKA-12557, there's a PR fixing it already.
--
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:
users@infra.apache.org
[GitHub] [kafka] vvcephei commented on a change in pull request #10378: KAFKA-7106: remove deprecated Windows APIs
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #10378: KAFKA-7106: remove deprecated Windows APIs
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #10378:
URL: https://github.com/apache/kafka/pull/10378#discussion_r602810827
##########
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:
When trimming the other inherited classes, I actually feel that some of the common fields like grace period may be better moved here in the future (I did not do this immediately to keep the PR small). WDYT?
--
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:
users@infra.apache.org
[GitHub] [kafka] guozhangwang merged pull request #10378: KAFKA-7106: remove deprecated Windows APIs
Posted by GitBox <gi...@apache.org>.
guozhangwang merged pull request #10378:
URL: https://github.com/apache/kafka/pull/10378
--
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:
users@infra.apache.org
[GitHub] [kafka] guozhangwang commented on a change in pull request #10378: KAFKA-7106: remove deprecated Windows APIs
Posted by GitBox <gi...@apache.org>.
guozhangwang commented on a change in pull request #10378:
URL: https://github.com/apache/kafka/pull/10378#discussion_r599306972
##########
File path: streams/src/test/java/org/apache/kafka/streams/processor/internals/RepartitionOptimizingTest.java
##########
@@ -185,8 +185,8 @@ private void runTest(final String optimizationConfig, final int expectedNumberRe
.filter((k, v) -> k.equals("A"), Named.as("join-filter"))
.join(countStream, (v1, v2) -> v1 + ":" + v2.toString(),
JoinWindows.of(ofMillis(5000)),
- StreamJoined.<String, String, Long>with(Stores.inMemoryWindowStore("join-store", ofDays(1), ofMillis(10000), true),
- Stores.inMemoryWindowStore("other-join-store", ofDays(1), ofMillis(10000), true))
+ StreamJoined.<String, String, Long>with(Stores.inMemoryWindowStore("join-store", ofDays(1).plus(ofMillis(10000)), ofMillis(10000), true),
Review comment:
This is the by-product of this change as we default grace to 24h, and retention to grace + window-size.
We can consider whether we keep the default grace at 24h cc @ableegoldman
##########
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 chose to use 24h, not 24h - size as the default value. Since it is simpler to set at construction time and I think the difference is insignificant. Ditto for other classes.
##########
File path: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregateTest.java
##########
@@ -317,18 +318,21 @@ public void shouldLogAndMeterWhenSkippingExpiredWindowWithBuiltInMetricsVersion0
shouldLogAndMeterWhenSkippingExpiredWindow(StreamsConfig.METRICS_0100_TO_24);
}
- @Deprecated // testing deprecated functionality (behavior of until)
private void shouldLogAndMeterWhenSkippingExpiredWindow(final String builtInMetricsVersion) {
final StreamsBuilder builder = new StreamsBuilder();
final String topic = "topic";
final KStream<String, String> stream1 = builder.stream(topic, Consumed.with(Serdes.String(), Serdes.String()));
stream1.groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
- .windowedBy(TimeWindows.of(ofMillis(10)).advanceBy(ofMillis(5)).until(100))
+ .windowedBy(TimeWindows.of(ofMillis(10)).advanceBy(ofMillis(5)))
.aggregate(
() -> "",
MockAggregator.toStringInstance("+"),
- Materialized.<String, String, WindowStore<Bytes, byte[]>>as("topic1-Canonicalized").withValueSerde(Serdes.String()).withCachingDisabled().withLoggingDisabled()
+ Materialized.<String, String, WindowStore<Bytes, byte[]>>as("topic1-Canonicalized")
Review comment:
I still maintained this test while moving the retention from window to materialized.
##########
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:
This class become a pure abstract one, and I'm moving the default value (also renamed it) from WindowsDefault and remove the other class.
##########
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:
This class is no more needed since Windows is a pure abstract class now.
--
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:
users@infra.apache.org