KAFKA-7277: Migrate Streams API to Duration instead of longMs times #5682

nizhikov · 2018-09-22T07:20:23Z

Public API changed according to KIP-358

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

nizhikov · 2018-09-22T07:34:10Z

Hello, @vvcephei
Please, give me an advice: Should I also migrate all calls of the deprecated method in tests?

…ew version called instead.

vvcephei

Hey @nizhikov ,

Thanks for the PR!

I've made a quick pass and left some comments that I think will apply to basically all of the methods, so I'll stop here.

At a high level, this is looking really good.

Please let me know if you have questions/objections to my feedback.

Thanks,
-John

vvcephei · 2018-09-24T20:33:05Z

streams/src/main/java/org/apache/kafka/streams/ApiUtils.java

+            throw new IllegalArgumentException(name + " shouldn't be null.", e);
+        } catch (final ArithmeticException e) {
+            throw new IllegalArgumentException(name + " can't be converted to milliseconds. " +  d +
+                " is negative or too big", e);


This validation doesn't actually check that it's non-negative. I'm not sure that we want to, though.
Also, I'd use Objects.toString() on name just in case it's null.
Finally, it might be nice to wrap the substituted variables with brackets to make the result more readable.
All together, this would yield:

"[" + Objects.toString(name) + "] shouldn't be null.", e);

"[" + Objects.toString(name) + "] is too big to be converted to milliseconds: [" + d + "]", e);

What do you think about this?
(also below)

vvcephei · 2018-09-24T20:33:44Z

streams/src/main/java/org/apache/kafka/streams/ApiUtils.java

+     * @param d Duration to check
+     * @param name Name of params for an error message.
+     */
+    public static void validateMillisecondDuration(final Duration d, final String name) {


Please avoid single-letter variables: d -> duration
(also below)

...ams/examples/src/main/java/org/apache/kafka/streams/examples/pageview/PageViewTypedDemo.java

vvcephei · 2018-09-24T21:24:53Z

...amples/src/main/java/org/apache/kafka/streams/examples/wordcount/WordCountProcessorDemo.java

@@ -63,7 +64,7 @@
                @SuppressWarnings("unchecked")
                public void init(final ProcessorContext context) {
                    this.context = context;
-                    this.context.schedule(1000, PunctuationType.STREAM_TIME, timestamp -> {
+                    this.context.schedule(Duration.ofMillis(1000), PunctuationType.STREAM_TIME, timestamp -> {


Feel free to add a static import for ofMillis and STREAMS_TIME to shorten this line.

vvcephei · 2018-09-24T21:27:40Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+     * @throws IllegalArgumentException if {@param timeout} is negative or too big
+     */
+    public synchronized boolean close(final Duration timeout) throws IllegalArgumentException {
+        log.debug("Stopping Streams client with timeoutMillis = {} ms.", timeout.toMillis());


We should use validateMillisecondDuration here.

vvcephei · 2018-09-24T21:52:15Z

streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java

+     * the timestamp of the record from the primary stream.
+     *
+     * @param timeDifference join window interval
+     * @throws IllegalArgumentException if {@code timeDifference} is negative or too big


As I mentioned before, we don't throw IAE if timeDifference is negative.
(also below)

vvcephei · 2018-09-24T21:54:11Z

streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java

@@ -109,11 +110,27 @@ private JoinWindows(final long beforeMs,
     * the timestamp of the record from the primary stream.
     *
     * @param timeDifferenceMs join window interval in milliseconds
-     * @throws IllegalArgumentException if {@code timeDifferenceMs} is negative
+     * @throws IllegalArgumentException if {@code timeDifferenceMs} is negative or too big


This method is actually guaranteed not to throw this exception. Since the parameter is a long of milliseconds, it's not possible to pass a value that is not expressible in milliseconds as a long.
(also applies elsewhere)

nizhikov · 2018-09-25T17:25:26Z

@vvcephei I refactor new methods. So they now call deprecated methods.
Please, review.

vvcephei · 2018-09-25T23:45:16Z

I'll take a look tomorrow. Thanks @nizhikov .

mjsax · 2018-09-26T00:37:42Z

streams/src/main/java/org/apache/kafka/streams/ApiUtils.java

+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams;


This is a public package and thus would make this class public API -- don't think we want this (also not part of the KIP). Maybe, we should create a new package org.apache.kafka.streams.internals and move it there?

streams/src/main/java/org/apache/kafka/streams/ApiUtils.java

mjsax · 2018-09-26T00:41:46Z

streams/src/main/java/org/apache/kafka/streams/ApiUtils.java

+
+/**
+ */
+public final class ApiUtils {


I am wondering, what we gain by introducing this class?

Hello @mjsax

As discussed in KIP thread we want to wrap NPE and ArithmeticException into IllegalArgumentException

See @vvcephei suggestion: "I still feel that surfacing the ArithmeticException directly would not be a great experience, so I still advocate for wrapping it in an IllegalArgumentException that explains our upper bound for Duration is "max-long number of milliseconds"

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

* ApiUtils moved to internals package. * JavaDoc fixes.

nizhikov · 2018-09-27T08:32:57Z

Hello @mjsax @vvcephei. I've fixed all your comments. Please, review.

vvcephei

Hi @nizhikov ,
I made another pass over this.

Mostly minor comments, with the exception of the one on streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java

Thanks again,
-John

streams/src/main/java/org/apache/kafka/streams/internals/ApiUtils.java

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java

streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java

streams/src/main/java/org/apache/kafka/streams/state/Stores.java

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoin.java

* Typos fixed. * Error messages improved. * Wrong new method removed * fetch, fetchAll added to WindowStore

nizhikov · 2018-09-27T22:45:13Z

Hello @vvcephei.
Thanks, for review.
I've fixed all your comments.

vvcephei · 2018-09-28T02:01:26Z

Failure unrelated.

Retest this, please.

nizhikov · 2018-09-28T04:35:58Z

Failure unrelated.

Retest this, please.

nizhikov · 2018-09-28T09:56:41Z

@vvcephei @mjsax I've checked failed tests locally and it all succeed.
Seems it some issue with test environment.
Please, give me an advice, what should I do?

dragon:~/src/kafka:[KAFKA-7277]$ scala -version
Scala code runner version 2.12.3 -- Copyright 2002-2017, LAMP/EPFL and Lightbend, Inc.
dragon:~/src/kafka:[KAFKA-7277]$ ./gradlew core:test --tests kafka.server.epoch.EpochDrivenReplicationProtocolAcceptanceTest.offsetsShouldNotGoBackwards


> Configure project :
Building project 'core' with Scala version 2.11.12
Building project 'streams-scala' with Scala version 2.11.12

> Task :core:test

kafka.server.epoch.EpochDrivenReplicationProtocolAcceptanceTest > offsetsShouldNotGoBackwards PASSED

BUILD SUCCESSFUL in 1m 15s
12 actionable tasks: 2 executed, 10 up-to-date
dragon:~/src/kafka:[KAFKA-7277]$ ./gradlew core:integrationTest --tests kafka.api.SaslSslAdminClientIntegrationTest.testMinimumRequestTimeouts

> Configure project :
Building project 'core' with Scala version 2.11.12
Building project 'streams-scala' with Scala version 2.11.12

> Task :core:integrationTest

kafka.api.SaslSslAdminClientIntegrationTest > testMinimumRequestTimeouts PASSED

BUILD SUCCESSFUL in 11s
10 actionable tasks: 2 executed, 8 up-to-date
dragon:~/src/kafka:[KAFKA-7277]$ ./gradlew core:test --tests kafka.admin.ConfigCommandTest.shouldExitWithNonZeroStatusOnArgError

> Configure project :
Building project 'core' with Scala version 2.11.12
Building project 'streams-scala' with Scala version 2.11.12

> Task :core:test

kafka.admin.ConfigCommandTest > shouldExitWithNonZeroStatusOnArgError PASSED

BUILD SUCCESSFUL in 2s
12 actionable tasks: 2 executed, 10 up-to-date

nizhikov · 2018-09-28T13:34:44Z

Failure unrelated.

Retest this, please.

nizhikov · 2018-09-29T05:47:48Z

Failure unrelated.

Retest this, please.

nizhikov · 2018-09-29T08:17:31Z

@vvcephei @mjsax Finally, tests passed. Please, review.

vvcephei

Thanks for the update @nizhikov ,

I have just one more request (which applies multiple times); otherwise, it looks good to me!

vvcephei · 2018-09-29T15:54:47Z

streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java

@@ -179,6 +182,7 @@ public synchronized void put(final Bytes key, final byte[] value, final long win
    }

    @Override
+    @Deprecated


These don't need to be marked deprecated, thanks to the non-deprecated definition in WindowStore now.

This comment applies to all the implementations of WindowStore in this PR.

@deprecated

* Unneeded @deprecated removed

vvcephei

Looks good to me!

Thanks @nizhikov

nizhikov · 2018-10-01T17:48:12Z

@mjsax I don't want to waste your time :)
Can you share your IDE config options regarding code style?
In Apache Ignite we have done this in wiki

nizhikov · 2018-10-01T20:01:47Z

@mjsax Tests passed. What else should be done for merge?

bbejeck

Thanks @nizhikov for the patch. I left some general comments but overall LGTM.

bbejeck · 2018-10-01T21:13:46Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+    /**
+     * Shutdown this {@code KafkaStreams} by signaling all the threads to stop, and then wait up to the timeout for the
+     * threads to join.
+     * A {@code timeout} of 0 means to wait forever.


Overall I'm ok with the semantics.

But my first instinct of a timeout of 0 implies shutdown immediately with no wait and blocking forever takes a value of Long.MAX_VALUE.

In short, I'm +1 as well on revising the behavior.

bbejeck · 2018-10-01T22:10:55Z

streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java

@@ -361,7 +364,7 @@ public boolean conditionMet() {
                        final int index = metadata.hostInfo().port();
                        final KafkaStreams streamsWithKey = streamRunnables[index].getStream();
                        final ReadOnlyWindowStore<String, Long> store = streamsWithKey.store(storeName, QueryableStoreTypes.<String, Long>windowStore());
-                        return store != null && store.fetch(key, from, to) != null;
+                        return store != null && store.fetch(key, ofEpochMilli(from), ofMillis(to - from)) != null;


For me, I have a slight preference for the from - to approach, but that's due more to personal preference.
I agree with @vvcephei that either way it's the same semantics just expressed differently.

So overall I'd prefer Instant, Instant, but if others want Instant, Duration I'm good with that as well.

nizhikov · 2018-10-02T16:41:08Z

I've resolved merge conflicts.

nizhikov · 2018-10-02T22:51:35Z

Failure unrelated.

Retest this, please.

mjsax · 2018-10-03T23:01:42Z

streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyWindowStore.java

+     * @throws NullPointerException If {@code null} is used for any key.
+     * @throws IllegalArgumentException if duration is negative or can't be represented as {@code long milliseconds}
+     */
+    KeyValueIterator<Windowed<K>, V> fetch(K from, K to, Instant fromTime, Duration duration)


I just talked to @guozhangwang @bbejeck and @vvcephei -- it seems, we are in favor to go with Instant/Instant types here and also for fetchAll(). 2.1 branch is not cut, and thus we can still get this into 2.1 -- can you update the PR accordingly? (Sorry for the long delay, but we were quite busy with many other deadline related things...)

@mjsax

I've updated (KIP-358)[https://cwiki.apache.org/confluence/display/KAFKA/KIP-358%3A+Migrate+Streams+API+to+Duration+instead+of+long+ms+times] and PR according to your proposal.

Please, review.

mjsax · 2018-10-03T23:03:42Z

streams/src/test/java/org/apache/kafka/streams/integration/QueryableStateIntegrationTest.java

@@ -361,7 +364,7 @@ public boolean conditionMet() {
                        final int index = metadata.hostInfo().port();
                        final KafkaStreams streamsWithKey = streamRunnables[index].getStream();
                        final ReadOnlyWindowStore<String, Long> store = streamsWithKey.store(storeName, QueryableStoreTypes.<String, Long>windowStore());
-                        return store != null && store.fetch(key, from, to) != null;
+                        return store != null && store.fetch(key, ofEpochMilli(from), ofMillis(to - from)) != null;


Just synced up with the others -- seems we are in favor to switch to Instant/Instant and try to get it into 2.1 (see my other comment).

nizhikov · 2018-10-04T14:35:08Z

@mjsax @bbejeck
I migrated fetch methods to Instant, Instant.
Tests passed.
Please, review.

vvcephei · 2018-10-04T14:47:32Z

Hey @nizhikov , I apologise that I didn't think of this earlier, but can you provide default implementations of the new fetch methods in the WindowStore interface? The default implementations would just defer to the long/long methods.

This would mean that any implementations would just automatically work with the new APIs. In fact, you could then pull the implementations out of all our window store implementations.

Does that make sense?

(I'm tempted not to bring it up at this late stage, but I think it would be very nice for store implementers not to have to add the trivial delegate methods.)

nizhikov · 2018-10-04T14:56:55Z

@vvcephei Do you mean we need to add default implementations into ReadOnlyWindowStore?

vvcephei · 2018-10-04T15:25:01Z

No, I mean in WindowStore (the Instant/Instant methods are available there, since WindowStore extends ReadOnlyWindowStore), and the long/long methods are not deprecated in this context.

nizhikov · 2018-10-04T15:27:42Z

@vvcephei

But we have classes that only extends ReadOnlyWindowStore. CompositeReadOnlyWindowStore, for example. So if we want to reduce code duplication we have to add default implementations to ReadOnlyWindowStore. Isn't it?

mjsax · 2018-10-04T15:30:55Z

@nizhikov @vvcephei

Because we deprecate ReadOnlyWindowStore#fetch(K, long, long) I don't think it makes sense to add default implementation there, that "map" from fetch(K, Instant, Instant) to #fetch(K, long, long)

However, I am not even sure how much we gain adding any default implementation. I am not against it, however, only people who implement their own WindowStore would be affected and those are very few... Might not be worth it.

nizhikov · 2018-10-04T15:34:56Z

@mjsax @vvcephei So maybe just merge it? :)

bbejeck

Thanks for the quick change @nizhikov. The changes LGTM

vvcephei · 2018-10-04T15:44:15Z

Sounds good to me.

mjsax · 2018-10-04T15:48:41Z

Cool. I'll wait until Jenkins passes (and go over the changes -- my initial "skip over review" looked good) and than merge, if @guozhangwang does not raise any concern.

vvcephei · 2018-10-04T16:14:21Z

Many thanks for this feature (and your patience) @nizhikov !

guozhangwang

Thanks for the contribution @nizhikov !

nizhikov · 2018-10-04T17:06:29Z

@mjsax Tests passed.

…pache#5682) Reviewers: Johne Roesler <[email protected]>, Matthias J. Sax <[email protected]>, Bill Bejeck <[email protected]>, Guozhang Wang <[email protected]>

nizhikov added 2 commits September 21, 2018 23:10

KAFKA-7277: Migration to Duration and Instant in public API.

0293466

Merge branch 'trunk' into KAFKA-7277

6724276

nizhikov added 3 commits September 23, 2018 00:28

KAFKA-7277: Call in tests and examples are changed to Duration versions.

9730461

KAFKA-7277: Call in tests and examples are changed to Duration versions.

c7b3c9a

KAFKA-7277: Tests fixed. All deprecated methods removed from tests, n…

c4f2335

…ew version called instead.

vvcephei reviewed Sep 24, 2018

View reviewed changes

nizhikov added 2 commits September 25, 2018 20:05

KAFKA-7277: All new method executed via deprecated.

a6d47d4

KAFKA-7277: Code review imrpovements

b44aed9

nizhikov added 2 commits September 26, 2018 00:56

Merge branch 'trunk' into KAFKA-7277

52c5a84

KAFKA-7277: Fixed compilation after merge with trunk.

4e42ed4

mjsax added the streams label Sep 26, 2018

mjsax reviewed Sep 26, 2018

View reviewed changes

KAFKA-7277: Code review fixes:

e5b9e5f

* ApiUtils moved to internals package. * JavaDoc fixes.

vvcephei requested changes Sep 27, 2018

View reviewed changes

KAFKA-7277: Code review fixes:

527f6fe

* Typos fixed. * Error messages improved. * Wrong new method removed * fetch, fetchAll added to WindowStore

vvcephei requested changes Sep 29, 2018

View reviewed changes

nizhikov added 2 commits September 29, 2018 19:53

Merge branch 'trunk' into KAFKA-7277

6b25162

KAFKA-7277: Code review fixes:

d63b729

* Unneeded @deprecated removed

vvcephei approved these changes Sep 29, 2018

View reviewed changes

bbejeck approved these changes Oct 1, 2018

View reviewed changes

Merge branch 'trunk' into KAFKA-7277

518a914

Merge branch 'trunk' of github.com:apache/kafka into KAFKA-7277

6eb48e3

mjsax reviewed Oct 3, 2018

View reviewed changes

nizhikov added 4 commits October 4, 2018 06:45

Merge branch 'trunk' of github.com:apache/kafka into KAFKA-7277

f241d3a

KAFKA-7277: Duration -> Instant in ReadOnlyWindowStore.

4e8b65f

KAFKA-7277: Duration -> Instant in ReadOnlyWindowStore. Tests passed.

a583259

KAFKA-7277: Javadoc fixed.

5210f9f

bbejeck approved these changes Oct 4, 2018

View reviewed changes

guozhangwang approved these changes Oct 4, 2018

View reviewed changes

mjsax merged commit ca641b3 into apache:trunk Oct 4, 2018

nizhikov mentioned this pull request Oct 5, 2018

KAFKA-7477: Improve Streams close timeout semantics #5747

Merged

3 tasks

vvcephei mentioned this pull request Oct 8, 2018

KAFKA-7277: default implementation for new window store overloads #5759

Merged

3 tasks

jeqo mentioned this pull request Aug 24, 2020

KAFKA-9929: Support backward iterator on WindowStore #9138

Merged

3 tasks

KAFKA-7277: Migrate Streams API to Duration instead of longMs times #5682

KAFKA-7277: Migrate Streams API to Duration instead of longMs times #5682

Conversation

nizhikov commented Sep 22, 2018

Committer Checklist (excluded from commit message)

nizhikov commented Sep 22, 2018

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nizhikov commented Sep 25, 2018

vvcephei commented Sep 25, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nizhikov commented Sep 27, 2018

vvcephei left a comment

Choose a reason for hiding this comment

nizhikov commented Sep 27, 2018

vvcephei commented Sep 28, 2018

nizhikov commented Sep 28, 2018

nizhikov commented Sep 28, 2018

nizhikov commented Sep 28, 2018

nizhikov commented Sep 29, 2018

nizhikov commented Sep 29, 2018

vvcephei left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei left a comment

Choose a reason for hiding this comment

nizhikov commented Oct 1, 2018 • edited Loading

nizhikov commented Oct 1, 2018

bbejeck left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nizhikov commented Oct 2, 2018

nizhikov commented Oct 2, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nizhikov commented Oct 4, 2018

vvcephei commented Oct 4, 2018 • edited Loading

nizhikov commented Oct 4, 2018

vvcephei commented Oct 4, 2018 • edited Loading

nizhikov commented Oct 4, 2018

mjsax commented Oct 4, 2018 • edited Loading

nizhikov commented Oct 4, 2018

bbejeck left a comment

Choose a reason for hiding this comment

vvcephei commented Oct 4, 2018

mjsax commented Oct 4, 2018

vvcephei commented Oct 4, 2018

guozhangwang left a comment

Choose a reason for hiding this comment

nizhikov commented Oct 4, 2018

nizhikov commented Oct 1, 2018 •

edited

Loading

vvcephei commented Oct 4, 2018 •

edited

Loading

vvcephei commented Oct 4, 2018 •

edited

Loading

mjsax commented Oct 4, 2018 •

edited

Loading