KAFKA-5233 KIP-138: Change punctuate semantics #3055

mihbor · 2017-05-15T14:37:17Z

Implementation for KIP-138: Change punctuate semantics

asfbot · 2017-05-15T14:40:52Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3914/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-05-15T14:41:09Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3925/
Test FAILed (JDK 7 and Scala 2.11).

asfbot · 2017-05-15T15:12:07Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3927/
Test FAILed (JDK 7 and Scala 2.11).

asfbot · 2017-05-15T15:43:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3917/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-05-15T16:10:36Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3928/
Test PASSed (JDK 7 and Scala 2.11).

enothereska · 2017-05-16T08:20:41Z

retest this please

asfbot · 2017-05-16T09:39:38Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3984/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-16T11:15:22Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/3987/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-16T11:26:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3975/
Test PASSed (JDK 8 and Scala 2.12).

enothereska · 2017-05-18T16:55:49Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

        if (processorContext.currentNode() != null) {
            throw new IllegalStateException(String.format("%s Current node is not null", logPrefix));
        }

        updateProcessorContext(new StampedRecord(DUMMY_RECORD, timestamp), node);

-        log.trace("{} Punctuating processor {} with timestamp {}", logPrefix, node.name(), timestamp);
+        log.trace("{} Punctuating processor {} with timestamp {} {}", logPrefix, node.name(), timestamp, type);


Nit: "with timestamp {} and type {}"

enothereska · 2017-05-18T17:01:57Z

Did a first pass over interfaces and depreciated functions, looks good. Haven't covered tests yet. FYI. @mihbor any chance you could rebase to latest trunk? Thanks.

enothereska · 2017-05-19T09:20:40Z

@mihbor a key thing that seems to be missing seems to be that if no events arrive, the system timer will never trigger. So we have again the issue that in absence of events, punctuate is never called. I could be missing something. Is the way around this to call maybePunctuate in the runLoop method in StreamThread.java right before calling maybeCommit? That way we can check the time after each poll. cc @mjsax @guozhangwang on this as well. This, and associated tests around system time, seem to be key things missing IMO. Thanks.

mjsax

I agree with @enothereska about missing improvements on when to punctuate.

mjsax · 2017-05-24T23:40:23Z

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

@@ -82,6 +83,7 @@
    R transform(final K key, final V value);

    /**
+     * <b>Deprecated as of 0.11.0.0</b> - <i>Please use {@link Punctuator)} functional interface instead.</i>


We should add @Deprecated Java annotation, and also use JavaDoc tag @Deprecated

mjsax · 2017-05-24T23:42:38Z

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

@@ -44,11 +44,13 @@
    void process(K key, V value);

    /**
+     * <b>Deprecated as of 0.11.0.0</b> - <i>Please use {@link Punctuator)} functional interface instead.</i>


Please use JavaDoc tag @Deprecated instead

mjsax · 2017-05-25T00:34:32Z

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

+     * schedule a periodic call called a punctuation to {@link Punctuator#punctuate(long)}.
+     *
+     * @param interval the time interval between punctuations
+     * @param type one of:<ul>>


nit: markup, two >>

I would also prefer to have this explanation in the body and not in the parameter list.

mjsax · 2017-05-25T00:35:51Z

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

+     *             in accordance with the timestamp as extracted by the {@link TimestampExtractor} in use.
+     *             <b>NOTE:</b> Only advanced if messages arrive</li>
+     *             <li>{@link PunctuationType#SYSTEM_TIME} - uses the system time (aka wall-clock time),
+     *             which is advanced at the polling interval ({@link org.apache.kafka.streams.StreamsConfig#POLL_MS_CONFIG})


advanced at the polling interval: not sure if this is true.

mjsax · 2017-05-25T00:37:05Z

...s/src/main/java/org/apache/kafka/streams/processor/internals/GlobalProcessorContextImpl.java

    @Override
    public <K, V> void forward(K key, V value, int childIndex) {
-        throw new UnsupportedOperationException();
+        throw new UnsupportedOperationException("this should not happen: forward() not supported in global processor context.");


Thanks for the cleanup! Appreciate it!

mjsax · 2017-05-25T00:40:19Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java

-        task.schedule(interval);
+        schedule(interval, PunctuationType.STREAM_TIME, new Punctuator() {
+            @Override
+            public void punctuate(long timestamp) {


nit: add final

mjsax · 2017-05-25T00:40:31Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java

@@ -122,8 +125,19 @@ public void commit() {
    }

    @Override
+    public Cancellable schedule(long interval, PunctuationType type, Punctuator callback) {


nit: add final to all parameters

mjsax · 2017-05-25T01:04:02Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java

@@ -422,12 +446,14 @@ int numBuffered() {
    boolean maybePunctuate() {
        final long timestamp = partitionGroup.timestamp();

+        boolean punctuated = systemTimePunctuationQueue.mayPunctuate(System.currentTimeMillis(), PunctuationType.SYSTEM_TIME, this);


we should not call System.currentTimeMillis() directly, but use the tasks Time object -- otherwise, we cannot test anything.

With regard to high level comment, we should on do this call here anyway.

mihbor · 2017-05-27T10:45:35Z

Oh, I see your point about when to punctuate, thanks.
Admittedly, I was rushing to get the PR submitted before the cut-off. Now, since that ship has sailed I'll have a more thorough look and work on test coverage as well.

asfbot · 2017-05-27T11:38:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4485/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-27T12:07:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4471/
Test PASSed (JDK 8 and Scala 2.12).

mihbor · 2017-05-27T12:31:55Z

Not sure about metrics. Should there be one punctuate time sensor or one for stream time and a separate one for system time punctuations?

mihbor · 2017-05-27T14:00:42Z

My naive implementation of cancellable was quite badly broken as it wasn't pointing at the latest PunctuationSchedule object, but always only the first one. About to commit something more involved but functional.

asfbot · 2017-05-27T15:39:53Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4488/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-27T16:04:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4474/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-05-27T18:23:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4490/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-27T18:25:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4476/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-05-28T10:11:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4509/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-28T10:33:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4494/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-06-03T11:36:41Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4856/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-06-03T11:40:40Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4872/
Test PASSed (JDK 7 and Scala 2.11).

enothereska · 2017-06-05T18:48:29Z

Thanks, changes LGTM. Calling for second reviews @dguy @bbejeck

mjsax

Some nits. Overall LGTM.

mjsax · 2017-06-05T19:35:48Z

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

@@ -28,8 +30,8 @@
 * This is a stateful record-by-record operation, i.e, {@link #transform(Object, Object)} is invoked individually for
 * each record of a stream and can access and modify a state that is available beyond a single call of
 * {@link #transform(Object, Object)} (cf. {@link KeyValueMapper} for stateless record transformation).
- * Additionally, the interface can be called in regular intervals based on the processing progress
- * (cf. {@link #punctuate(long)}.
+ * Additionally, this Transformer can {@link ProcessorContext#schedule(long, PunctuationType, Punctuator) schedule}


Nit: {@code Transformer}

mjsax · 2017-06-05T19:36:45Z

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

@@ -28,8 +30,8 @@
 * This is a stateful record-by-record operation, i.e, {@link #transform(Object)} is invoked individually for each
 * record of a stream and can access and modify a state that is available beyond a single call of
 * {@link #transform(Object)} (cf. {@link ValueMapper} for stateless value transformation).
- * Additionally, the interface can be called in regular intervals based on the processing progress
- * (cf. {@link #punctuate(long)}.
+ * Additionally, this ValueTransformer can {@link ProcessorContext#schedule(long, PunctuationType, Punctuator) schedule}


Nit: {@code ValueTransformer}

mjsax · 2017-06-05T19:38:21Z

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

@@ -82,6 +84,7 @@
    R transform(final K key, final V value);

    /**
+     * @deprecated As of 0.11.1.0 please use {@link Punctuator} functional interface instead.


Nit: remove As of 0.11.1.0 -- we also don't know yet what the next version number will be. If it's not 0.11.1.0 it would be wrong and I am pretty sure we will miss to update it.

mjsax · 2017-06-05T19:39:09Z

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

@@ -82,6 +84,7 @@
    VR transform(final V value);

    /**
+     * @deprecated As of 0.11.1.0 please use {@link Punctuator} functional interface instead.


mjsax · 2017-06-05T19:39:30Z

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

@@ -44,11 +45,13 @@
    void process(K key, V value);

    /**
+     * @deprecated As of 0.11.1.0 please use {@link Punctuator} functional interface instead.


mjsax · 2017-06-05T19:39:43Z

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

+    Cancellable schedule(long interval, PunctuationType type, Punctuator callback);
+
+    /**
+     * @deprecated As of 0.11.1.0 please use {@link #schedule(long, PunctuationType, Punctuator)} instead.


mjsax · 2017-06-05T19:40:43Z

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

@@ -88,11 +88,35 @@

    /**
     * Schedules a periodic operation for processors. A processor may call this method during
+     * {@link Processor#init(ProcessorContext) initialization} or processing
+     * {@link Processor#process(Object, Object)} to
+     * schedule a periodic call called a punctuation to {@link Punctuator#punctuate(long)}.


nit: a periodic call—called a punctuation—to

mjsax · 2017-06-05T19:42:21Z

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

+package org.apache.kafka.streams.processor;
+
+/**
+ * A functional interface used as an argument to {@link ProcessorContext#schedule(long, PunctuationType, Punctuator)}


Super-nit: add . at the end

mihbor · 2017-06-23T08:17:26Z

Sure, I'll change it. I'm also not a fan of premature optimisation and I think the impact would be negligible, but I didn't want to depart from the current practice for handling delegates in that class.

…tests and fixes

mihbor · 2017-06-24T15:04:28Z

rebased

asfgit · 2017-06-24T16:00:47Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/5688/
Test PASSed (JDK 7 and Scala 2.11).

asfgit · 2017-06-24T16:23:11Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/5674/
Test PASSed (JDK 8 and Scala 2.12).

dguy

LGTM

dguy · 2017-06-28T10:26:58Z

Thanks @mihbor for your efforts on this KIP! Merged to trunk

peoplemerge · 2017-08-30T17:57:45Z

Thanks for this work, @mihbor

I added trunk dep to my code base, I found the interface change self-describing, even near the end of a 10 hour day coding. My code was:
this.context.schedule(150l);

and became:
this.context.schedule(150l, PunctuationType.SYSTEM_TIME, this);

It was with great relief I saw all of our tests magically flip from failure to pass. I requested a hotfix to KIP-138 for 0.11.0.0.

How's that for validation?

mihbor · 2017-08-31T08:12:38Z

Thanks for your kind comment :-)

Cherry-picker: Dave Thomas <david.thomas@ticketmaster.com> Add revisionId to gradle-props. KAFKA-5233; KIP-138: Change punctuate semantics Implementation for KIP-138: Change punctuate semantics Author: Michal Borowiecki <michal.borowiecki@openbet.com> Reviewers: Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bbejeck@gmail.com>, Eno Thereska <eno.thereska@gmail.com>, Damian Guy <damian.guy@gmail.com> Closes apache#3055 from mihbor/KIP-138

mihbor changed the title ~~Kip 138~~ KAFKA-5233 KIP-138 May 15, 2017

enothereska reviewed May 18, 2017

View reviewed changes

mjsax requested changes May 25, 2017

View reviewed changes

mihbor force-pushed the KIP-138 branch from 2f59b5a to b763457 Compare May 27, 2017 10:33

mjsax approved these changes Jun 5, 2017

View reviewed changes

Michal Borowiecki added 18 commits June 24, 2017 16:01

KAFKA-5233 KIP-138 WIP stream time only

cf6a0e5

KAFKA-5233 KIP-138 WIP added support for system time punctuation

2bfeabb

KAFKA-5233 KIP-138 WIP refactored

5890420

KAFKA-5233 KIP-138 WIP fixes

4ea166e

KAFKA-5233 KIP-138 WIP trivia

e50a30d

KAFKA-5233 KIP-138 checkstyle

ce8b5f5

KAFKA-5233 KIP-138 checkstyle tests

08d2309

KAFKA-5233 KIP-138 remove erroneously added test

5c0558a

KAFKA-5233 KIP-138 need to use unconditional or

00c312d

KAFKA-5233 KIP-138 javadoc improvements

61b1b9c

KAFKA-5233 KIP-138 add final modifier to added method params

8ece45f

KAFKA-5233 KIP-138 system time punctuation and schedule cancellation …

3e2ef17

…tests and fixes

KAFKA-5233 KIP-138 another javadoc tweak

26e53f4

KAFKA-5233 KIP-138 more javadoc tweaks

2e8616e

KAFKA-5233/KIP-138 updated docs

4f8fef5

KAFKA-5233 KIP-138 javadoc nitpicks

07d1cb8

KAFKA-5233/KIP-138 post-review actions

d54ece7

KAFKA-5233/KIP-138 post-review actions

457286d

mihbor force-pushed the KIP-138 branch from f61a93f to 457286d Compare June 24, 2017 15:03

dguy approved these changes Jun 28, 2017

View reviewed changes

asfgit closed this in efb060c Jun 28, 2017

peoplemerge mentioned this pull request Sep 13, 2017

KAFKA-5233 follow up #3678

Closed

KAFKA-5233 KIP-138: Change punctuate semantics #3055

KAFKA-5233 KIP-138: Change punctuate semantics #3055

Conversation

mihbor commented May 15, 2017 • edited Loading

asfbot commented May 15, 2017

asfbot commented May 15, 2017

asfbot commented May 15, 2017

asfbot commented May 15, 2017

asfbot commented May 15, 2017

enothereska commented May 16, 2017

asfbot commented May 16, 2017

asfbot commented May 16, 2017

asfbot commented May 16, 2017

Choose a reason for hiding this comment

enothereska commented May 18, 2017

enothereska commented May 19, 2017

mjsax 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mihbor commented May 27, 2017

asfbot commented May 27, 2017

asfbot commented May 27, 2017

mihbor commented May 27, 2017

mihbor commented May 27, 2017

asfbot commented May 27, 2017

asfbot commented May 27, 2017

asfbot commented May 27, 2017

asfbot commented May 27, 2017

asfbot commented May 28, 2017

asfbot commented May 28, 2017

asfbot commented Jun 3, 2017

asfbot commented Jun 3, 2017

enothereska commented Jun 5, 2017

mjsax 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mihbor commented Jun 23, 2017

mihbor commented Jun 24, 2017

asfgit commented Jun 24, 2017

asfgit commented Jun 24, 2017

dguy left a comment

Choose a reason for hiding this comment

dguy commented Jun 28, 2017

peoplemerge commented Aug 30, 2017 • edited Loading

mihbor commented Aug 31, 2017

mihbor commented May 15, 2017 •

edited

Loading

peoplemerge commented Aug 30, 2017 •

edited

Loading