KAFKA-10546: Deprecate old PAPI by vvcephei · Pull Request #10869 · apache/kafka

vvcephei · 2021-06-11T19:59:07Z

Deprecate the old Processor API
Suppress warnings on all internal usages of the old API
(which will be migrated in other child tickets of KAFKA-8410)
Add new KStream#process methods, since KAFKA-10603 has not seen any action.

Committer Checklist (excluded from commit message)

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

vvcephei

Do you mind taking a look at this?

The original plan was to migrate all the internal processors and then deprecate the old PAPI, but it seems like we may not make the 3.0 release feature freeze with that approach. Instead, I'm proposing this PR to go ahead and deprecate the old API first and just suppress all deprecation warnings on our internals.

There was one side-effect. If you look at KIP-478, you'll see we proposed not to change the KStream#process method, in the hopes that we would already have a good replacement by the time of the deprecation. It also doesn't seem like that's going to happen, so I'm going to update the proposal to actually deprecate those methods and add new-PAPI versions.

Despite there being a lot of changed lines in this PR, I think it should be a pretty quick read. Most of the changes are:

suppressing warnings
switching to fully-qualified class names for the old API, since it's not possible to suppress deprecation warnings on imports.

vvcephei · 2021-06-11T20:02:44Z

     * Adds a state store to the underlying {@link Topology}.
     * <p>
-     * It is required to connect state stores to {@link org.apache.kafka.streams.processor.Processor Processors}, {@link Transformer Transformers},
+     * It is required to connect state stores to {@link org.apache.kafka.streams.processor.api.Processor Processors},


Switching to the new interface where possible.

vvcephei · 2021-06-11T20:03:14Z

     *
     * @param name the unique name of the processor node
-     * @param supplier the supplier used to obtain this node's {@link Processor} instance
+     * @param supplier the supplier used to obtain this node's {@link org.apache.kafka.streams.processor.Processor} instance


Specifying old PAPI by fully qualified name where necessary.

vvcephei · 2021-06-11T20:04:27Z

     * <p>
     * Note that it is possible to emit multiple records for each input record by using
-     * {@link ProcessorContext#forward(Object, Object) context#forward()} in
+     * {@link org.apache.kafka.streams.processor.ProcessorContext#forward(Object, Object) context#forward()} in


Transformation will be migrated separately via https://issues.apache.org/jira/browse/KAFKA-10603, so we just need to make sure the javadoc references the correct context. Note that for this reason, we did not deprecate the old ProcessorContext.

vvcephei · 2021-06-11T20:09:48Z

+     * @deprecated Since 3.0. Use {@link KStream#process(org.apache.kafka.streams.processor.api.ProcessorSupplier, java.lang.String...)} instead.
+     */
+    @Deprecated
+    void process(final org.apache.kafka.streams.processor.ProcessorSupplier<? super K, ? super V> processorSupplier,
+                 final String... stateStoreNames);


Note, git got confused. This is the pre-existing API, which is deprecated now.

vvcephei · 2021-06-11T20:11:21Z

+    <KOut, VOut> void process(final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+                              final String... stateStoreNames);


Here is the new API. Note: the forwarding type should technically be Void, since you can't forward from these processors, but I left it unbounded to potentially be compatible with whatever we decide to do in https://issues.apache.org/jira/browse/KAFKA-10603

Just wondering, if we ended up not doing KAFKA-10603, then how to bound KOut/Vout to Void only?

Thanks, @guozhangwang !

Yeah, it's kind of an either/or thing. If we make it KOut, VOut now, we can't tighten it to Void, Void later.

Although, coming back to it, I'm thinking that we actually should be able to make it Void, Void now and then relax it to KOut, VOut later. Maybe you can sanity-check me:

Now:

Def: void process(final ProcessorSupplier<? super K, ? super V, Void, Void> processorSupplier) Usage: mystream.process(new ProcessorSupplier<String, String, Void, Void>(){ ... Processor<String, String, Void, Void> get() { // calling forward is optional, but if they do it, they could only call: context.forward(new Record(null, null, ...)); } ... })

Later:

Def: KStream<KOut, VOut> process(final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier) Same Usage is still ok: * Return is unused, which is fine * Return KOut and VOut both bind to Void, which is also fine. mystream.process(new ProcessorSupplier<String, String, Void, Void>(){ ... Processor<String, String, Void, Void> get() { // calling forward is optional, but if they do it, they could only call: context.forward(new Record(null, null, ...)); } ... })

So it seems we actually could/should set them to Void right now. Does that seem right to you?

If any template type K could bind to Void then that seems fine -- I was not clear about this myself :P

@vvcephei +1 I like the last proposal: using Void and keep the option to migrate to K/V later.

vvcephei · 2021-06-11T20:24:21Z

        final NullPointerException exception = assertThrows(
            NullPointerException.class,
-            () -> testStream.process(null));
+            () -> testStream.process((ProcessorSupplier<? super String, ? super String, Object, Object>) null));


We have to cast null now to resolve the overload we want.

vvcephei · 2021-06-11T20:26:39Z

+  @deprecated(since = "3.0", message = "Use process(ProcessorSupplier, String*) instead.")
+  def process(processorSupplier: () => org.apache.kafka.streams.processor.Processor[K, V],
+              stateStoreNames: String*): Unit = {
+    val processorSupplierJ: org.apache.kafka.streams.processor.ProcessorSupplier[K, V] = () => processorSupplier()
    inner.process(processorSupplierJ, stateStoreNames: _*)
  }


This is the mirror of the Java API: we have to do the same thing in Scala for the same reason.

vvcephei · 2021-06-11T20:27:46Z

+   * Note that this overload takes a ProcessorSupplier instead of a Function to avoid post-erasure ambiguity with
+   * the older (deprecated) overload.


Note this unfortunate fact. This means that Scala users won't be able to use SAM conversion anymore for this particular method. The only alternative we have is to give the new methods a different name.

vvcephei · 2021-06-11T20:29:02Z

+  Reducer,
+  Transformer,
+  ValueJoiner,
+  ValueMapper,


For whatever reason, Spotless Scala insisted on changing the order.

vvcephei · 2021-06-11T20:30:00Z

+  private class SimpleProcessorSupplier private[TopologyTest] (val valueList: util.List[String])
+      extends ProcessorSupplier[String, String, Nothing, Nothing] {
+
+    override def get(): Processor[String, String, Nothing, Nothing] =
+      (record: api.Record[String, String]) => valueList.add(record.value())


I went ahead and changed the test to demonstrate that SAM conversion doesn't work with these methods, so we have to declare a supplier instead of a () => processor.

vvcephei · 2021-06-11T20:44:09Z

Follow-on to #10507

Again, trying to help break up #10381

vvcephei · 2021-06-11T20:50:09Z

We'll have to take a look at the scala 2.12 builds, since I don't think the suppression I used is available in that version.

…tream.process methods, which means I also have to add a bunch more suppressions where that method is used in tests.

guozhangwang

I made a quick pass on the public classes, just one question regarding forward-looking for https://issues.apache.org/jira/browse/KAFKA-10603.

guozhangwang · 2021-06-13T05:09:25Z

+    <KOut, VOut> void process(final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
+                              final String... stateStoreNames);


Just wondering, if we ended up not doing KAFKA-10603, then how to bound KOut/Vout to Void only?

jeqo

LGTM

vvcephei · 2021-06-18T14:42:59Z

Thanks for the review, @jeqo !

I think we're pretty much good to go on this, once @guozhangwang has a chance to review the recent changes to the KStream#process methods.

guozhangwang

Thanks!

guozhangwang · 2021-06-19T16:04:23Z

@vvcephei I took a quick look at the more recent commits (for KStream.java the part you wanted me to take another look is around the javadoc copy-paste errors, right?), still looks good to me.

vvcephei · 2021-06-22T14:15:11Z

Thanks, @guozhangwang ! Yes, that's the area of the code I was looking for a double-check on. I switched the KOut/VOut parameters to Void/Void as you suggested.

* Deprecate the old Processor API * Suppress warnings on all internal usages of the old API (which will be migrated in other child tickets of KAFKA-8410) * Add new KStream#process methods, since KAFKA-10603 has not seen any action.

John Roesler added 8 commits June 11, 2021 11:37

first pass

2196b2d

checkstyle forces us to use FQNs for deprecated APIs

d62d82d

spotless

41755f1

can't import deprecated classes

193b691

spotless

443179a

suppress warnings

870d207

cleanup

5e60738

cleanup

c261a33

vvcephei requested a review from cadonna June 11, 2021 20:01

vvcephei self-assigned this Jun 11, 2021

vvcephei added the streams label Jun 11, 2021

fix interface deprecation

e871030

vvcephei commented Jun 11, 2021

View reviewed changes

I realized I should actually deprecate instead of suppress the old KS…

04ab7bd

…tream.process methods, which means I also have to add a bunch more suppressions where that method is used in tests.

guozhangwang reviewed Jun 13, 2021

View reviewed changes

John Roesler added 4 commits June 15, 2021 15:29

wip

d9c19db

fix scala KStream API

da30f69

fix

a0d46db

fix

a77cf40

jeqo reviewed Jun 17, 2021

View reviewed changes

Comment thread streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java Outdated

jeqo approved these changes Jun 17, 2021

View reviewed changes

Fix copy/paste error

a33ae96

guozhangwang approved these changes Jun 18, 2021

View reviewed changes

vvcephei merged commit c347508 into apache:trunk Jun 22, 2021

vvcephei deleted the deprecate-old-papi branch June 22, 2021 14:17

		<KOut, VOut> void process(final ProcessorSupplier<? super K, ? super V, KOut, VOut> processorSupplier,
		final String... stateStoreNames);

		* Note that this overload takes a ProcessorSupplier instead of a Function to avoid post-erasure ambiguity with
		* the older (deprecated) overload.

Conversation

vvcephei commented Jun 11, 2021

Committer Checklist (excluded from commit message)

Uh oh!

vvcephei left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vvcephei commented Jun 11, 2021

Uh oh!

vvcephei commented Jun 11, 2021

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

jeqo left a comment

Choose a reason for hiding this comment

Uh oh!

vvcephei commented Jun 18, 2021

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jun 19, 2021

Uh oh!

vvcephei commented Jun 22, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants