KAFKA-8410: Part 1: processor context bounds #8414

vvcephei · 2020-04-03T04:01:53Z

Add type bounds to the ProcessorContext, which bounds the types that can be forwarded to child nodes.

Committer Checklist (excluded from commit message)

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

vvcephei · 2020-04-03T04:02:25Z

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

@@ -64,7 +64,7 @@
     *
     * @param context the context
     */
-    void init(final ProcessorContext context);
+    void init(final ProcessorContext<Object, Object> context);


In the KIP, I proposed to leave the existing Processor and Transformer init methods with a raw ProcessorContext argument. This will give all users a rawtypes warning with no recourse. Instead, in this PR, I'm adding <Object, Object> bounds instead. This at least allows users to also add the generics if they desire, while not changing the practical bounds at all. If this proves acceptable, I'll update the KIP.

Not sure if I understand this. Can you elaborate?

Btw: to we think we will have any path forward to let user specify the actually return type (that is currently only R but does not specify the actually key or value type), such that we would pass in ProcessorContext<K, V>? Can't remember if we discussed this on the KIP (just a side question).

Good question. Yes, it's been forever, but we did (myself, I had to look back over the KIP to figure this out when I wrote the comment).

The idea is to discuss separately how to migrate to a version of Transformer that allows bounding the output types. The ticket for this is https://issues.apache.org/jira/browse/KAFKA-8396. This KIP was already sweeping enough in its design and implementation scope that we wanted to give Transformer its own discussion. Accordingly, this KIP only has to deal with the Processor API change and the consequences that are necessary/convenient to tackle at the same time.

Part of the reason to give Transformer its own discussion is that there are other opportunities to simplify transform() use cases (see the ticket for more info), so it really would be its own conversation.

What I was talking about above is that I proposed in this KIP only to leave Transformer completely alone, so this would still be a raw type: void init(final ProcessorContext context). But what I realized just now is that anyone who implements the interface will get a "raw types" warning from javac, and there is nothing they can do about it but ignore it. If they try to add the equivalent <Object, Object> or <?, ?> parameters, their implementation would clash with our interface.

OTOH, if we go ahead and add the parameters as I proposed above, then at least they can also add the <Object, Object> parameters to their code to resolve the warning. They can not bound the types tighter than that, and neither can we, until we pick up KAFKA-8396. So, it doesn't provide any additional type safety, just a way to resolve the warning without resolving to ignoring it. AFAICT, there is no way to compatibly migrate the interface to accommodate greater type safety. KAFKA-8396 will have to include deprecating something in any case.

Thanks for clarification.

vvcephei · 2020-04-03T04:02:48Z

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

@@ -69,7 +69,7 @@
     * @throws IllegalStateException If store gets registered after initialization is already finished
     * @throws StreamsException if the store's change log does not contain the partition
     */
-    void init(final ProcessorContext context);
+    void init(final ProcessorContext<Void, Void> context);


As proposed, Void types indicate that you can't forward anything.

vvcephei · 2020-04-03T04:04:27Z

...ams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java

        super.init(context);
        metrics = (StreamsMetricsImpl) context.metrics();
        droppedRecordsSensor = droppedRecordsSensorOrSkippedRecordsSensor(Thread.currentThread().getName(), context.taskId().toString(), metrics);
-        valueGetter.init(context);
+        valueGetter.init(new ForwardingDisabledProcessorContext(context));


Even in this first PR, we expose some bugs. In this case, ValueGetters and ValueTransformers shouldn't be able to forward anything, but nothing previously prevented them from doing it. Now, the new bounds force us to wire in a context that has Void bounds.

vvcephei · 2020-04-03T04:10:20Z

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

-    private final InternalProcessorContext context;
-    private final ProcessorNode myNode;
+    private final InternalProcessorContext<Object, Object> context;
+    private final ProcessorNode<?, ?> myNode;


You'll notice that sometimes, I use wildcard (?) generic parameter bounds and other times, Object. The reason is a little subtle. It's because the InternalProcessorContext has a method that takes a ProcessorNode argument. We want to let (ourselves) pass in any ProcessorNode, not just an exactly <Object, Object> one. But for all the ProcessorContext references, we pass the context in, and we have to do it untyped anyway, so the type bound doesn't matter. It results in less mind-bending code for users to declare types using the Object,Object bounds, and it makes no difference to them, as far as the calls to forward, since the parameterized types are only upper bounds anyway.

Punchline, it doesn't really matter, but there are considerations that led me to use ? in some cases and Object in others.

vvcephei · 2020-04-03T04:13:20Z

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

     * @return a handle allowing cancellation of the punctuation schedule established by this method
     */
    Cancellable schedule(final Duration interval,
                         final PunctuationType type,
-                         final Punctuator callback) throws IllegalArgumentException;
+                         final Punctuator callback);


Not part of the KIP, and no difference whatsoever to the public API, but (unlike the javadoc I added), declaring that a method throws an unchecked exception is a totally ineffective way to inform users that it might get thrown (since they'll only ever see the declaraction if they read the source of this file). At least, the javadoc will show up in their IDE.

vvcephei · 2020-04-03T04:15:01Z

...ava/org/apache/kafka/streams/processor/internals/ForwardingDisabledProcessorContextTest.java

-import org.junit.runner.RunWith;
-
-@RunWith(EasyMockRunner.class)
-public class ForwardingDisabledProcessorContextTest {


This test doesn't even compile now, because the compiler checks all these constraints for us. There's no way to express the things we wanted to forbid.

vvcephei · 2020-04-03T20:30:17Z

Hmm, that last commit illustrates a problem with this approach. Adding generics to a previously raw type is only a compiler warning for Java users, but it's a compiler error for Scala users, because Scala insists on having the generic arguments for any type that takes generic parameters.

I think the way to fix this is to propose to add a new ProcessorContext as well as a new Processor interface. Or we can accept this level of source code incompatibility in the Scala API. For context, many changes we've made in minor versions break Scala source compatibility, since Scala generally makes it harder for libraries to implement non-breaking deprecation paths.

mjsax · 2020-04-04T02:30:04Z

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

-    public <K, V> void forward(final K key,
-                               final V value) {
+    public <K1 extends K, V1 extends V> void forward(final K1 key,
+                               final V1 value) {


nit: fix indention (similar below)

Ah, forgot to reformat. Thanks for the catch.

vvcephei · 2020-04-06T22:05:05Z

Hey, thanks for the review @mjsax ! I resolved the formatting problem and answered your question.

mjsax · 2020-04-07T04:17:28Z

Java 11: `kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
Java 8:

kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
kafka.api.PlaintextConsumerTest.testLowMaxFetchSizeForRequestAndPartition

Retest this please.

mjsax

LGTM. Feel free to merge after Jenkins passed.

vvcephei · 2020-04-07T18:09:35Z

Filed https://issues.apache.org/jira/browse/KAFKA-9831 for the unrelated failure.

vvcephei · 2020-04-07T18:09:44Z

Thanks, @mjsax !

The AK PR apache/kafka#8414 introduced type parameters for the ProcessorContext. Since Scala requires generic arguments for any type that takes generic parameters, the Scala code in the examples that uses the ProcessorContext did not compile anymore.

This reverts commit 29e08fd. There turned out to be more than expected problems with adding the generic parameters. Reviewers: Matthias J. Sax <matthias@confluent.io>

vvcephei commented Apr 3, 2020

View reviewed changes

vvcephei changed the title ~~Kafka 8410 1 processor context bounds~~ KAFKA-8410: Part 1: processor context bounds Apr 3, 2020

John Roesler added 2 commits April 3, 2020 15:30

KAKFA-8410: Part 1: Add bounds to ProcessorContext

4bcfe92

fix scala build

2a68011

mjsax reviewed Apr 4, 2020

View reviewed changes

formatting

0eb647f

mjsax added the streams label Apr 7, 2020

mjsax approved these changes Apr 7, 2020

View reviewed changes

vvcephei merged commit 29e08fd into apache:trunk Apr 7, 2020

vvcephei deleted the kafka-8410-1-ProcessorContext-bounds branch April 7, 2020 18:11

vvcephei mentioned this pull request Apr 7, 2020

KIP-557: Add Emit On Change Support #8254

Merged

3 tasks

cadonna mentioned this pull request Apr 23, 2020

KSTREAMS-4026: Fix type parameter bounds of ProcessorContext in Scala confluentinc/kafka-streams-examples#330

Closed

mjsax added the kip Requires or implements a KIP label Jun 12, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-8410: Part 1: processor context bounds #8414

KAFKA-8410: Part 1: processor context bounds #8414

vvcephei commented Apr 3, 2020

vvcephei Apr 3, 2020

mjsax Apr 4, 2020

vvcephei Apr 6, 2020

mjsax Apr 7, 2020

vvcephei Apr 3, 2020

vvcephei Apr 3, 2020

vvcephei Apr 3, 2020

vvcephei Apr 3, 2020

vvcephei Apr 3, 2020

mjsax Apr 4, 2020

vvcephei commented Apr 3, 2020

mjsax Apr 4, 2020

vvcephei Apr 6, 2020

vvcephei commented Apr 6, 2020

mjsax commented Apr 7, 2020

mjsax left a comment

vvcephei commented Apr 7, 2020

vvcephei commented Apr 7, 2020

KAFKA-8410: Part 1: processor context bounds #8414

KAFKA-8410: Part 1: processor context bounds #8414

Conversation

vvcephei commented Apr 3, 2020

Committer Checklist (excluded from commit message)

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

vvcephei commented Apr 3, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Apr 6, 2020

mjsax commented Apr 7, 2020

mjsax left a comment

Choose a reason for hiding this comment

vvcephei commented Apr 7, 2020

vvcephei commented Apr 7, 2020