KAFKA-12829: Remove deprecated Topology#addProcessor of old Processor API #18154

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Merged

mjsax merged 2 commits into apache:trunk from mjsax:kafka-12829-remove-topology-process

Dec 20, 2024

Member

mjsax commented Dec 12, 2024 •

edited

Loading

Must be cherry-picked to 4.0 branch.

mjsax added the streams label

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamKStreamJoinTest.java

@@ @@ -57,7 +57,7 @@ @@
               import org.apache.kafka.test.GenericInMemoryKeyValueStore;
               import org.apache.kafka.test.MockApiProcessor;
               import org.apache.kafka.test.MockApiProcessorSupplier;
-              import org.apache.kafka.test.MockInternalNewProcessorContext;
+              import org.apache.kafka.test.MockInternalProcessorContext;

Member Author

mjsax Dec 12, 2024

Follow up clean from #18103

Rename this class, what result in lot of updates in many files...

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamProcessorApiTest.java

@@ @@ -46,7 +46,7 @@ @@
               import static java.util.Arrays.asList;
-              public class KStreamNewProcessorApiTest {
+              public class KStreamProcessorApiTest {

Member Author

mjsax Dec 12, 2024

This is a similar update -- we don't have the old Processor API any longer, so dropping the New from this name, too.

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

    
                      props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

                      props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass().getName());

                      props.setProperty(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());

                      props.setProperty(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class.getName());

Member Author

mjsax Dec 12, 2024

Side cleanup to get rid of some IntelliJ warning. (also cleaner code.

Similar stuff below.

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

-                      assertEquals("value2", store.get("key2"));
-                      assertEquals("value3", store.get("key3"));
-                      assertNull(store.get("key4"));
-                  }

Member Author

mjsax Dec 12, 2024

Diff is just weird.. this was not removed

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

-                  @Deprecated // testing old PAPI
-                  @Test
-                  public void testDrivingConnectedStateStoreInDifferentProcessorsTopologyWithOldAPI() {

Member Author

mjsax Dec 12, 2024

removed this (and some other tests) -- verified that we have equivalent tests for new Processor API for all of them

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

-                      assertEquals("key3", results.get(2).key);
-                      assertEquals("value3", results.get(2).value);
-                  }

Member Author

mjsax Dec 12, 2024

Weird diff..

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

                               .addSink("sink", OUTPUT_TOPIC_1, new DroppingPartitioner(), "processor");
                   }
-                  @Deprecated // testing old PAPI

Member Author

mjsax Dec 12, 2024

This annotation was wrong -- should have been @SuppressWarning("deprecation")

Updating this code to use new Processor API now.

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

                           .addStateStore(Stores.keyValueStoreBuilder(Stores.inMemoryKeyValueStore(storeName), Serdes.String(), Serdes.String()), "processor")
                           .addSink("counts", OUTPUT_TOPIC_1, "processor");
                   }
-                  @Deprecated // testing old PAPI

Member Author

mjsax Dec 12, 2024

as above

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

-                   * A processor that stores each key-value pair in an in-memory key-value store registered with the context.
-                   */
-                  @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
-                  protected static class OldAPIStatefulProcessor extends org.apache.kafka.streams.processor.AbstractProcessor<String, String> {

Member Author

mjsax Dec 12, 2024

There is already a StatefulProcessor for the new PAPI -- nothing to be migrated -- deleting.

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

                       }
                   }
-                  @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
-                  private <K, V> org.apache.kafka.streams.processor.ProcessorSupplier<K, V> define(final org.apache.kafka.streams.processor.Processor<K, V> processor) {

Member Author

mjsax Dec 12, 2024

just dropped this one -- not really useful to have..

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

-                  }
-                  @SuppressWarnings("deprecation") // Old PAPI. Needs to be migrated.
-                  private <K, V> org.apache.kafka.streams.processor.ProcessorSupplier<K, V> defineWithStoresOldAPI(final Supplier<org.apache.kafka.streams.processor.Processor<K, V>> supplier,

Member Author

mjsax Dec 12, 2024

There is already defineWithStores() for new PAPI -- nothing to be migrated -- deleting

mjsax commented

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

                   private <KIn, VIn, KOut, VOut> ProcessorSupplier<KIn, VIn, KOut, VOut> defineWithStores(final Supplier<Processor<KIn, VIn, KOut, VOut>> supplier,
                                                                                                           final Set<StoreBuilder<?>> stores) {
-                      return new ProcessorSupplier<KIn, VIn, KOut, VOut>() {
+                      return new ProcessorSupplier<>() {

Member Author

mjsax Dec 12, 2024

side cleanup

mjsax mentioned this pull request

KAFKA-12829: Remove deprecated Topology#addProcessor of old Processor API #17190

Closed

3 tasks

pegasas reviewed

View reviewed changes

streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorTopologyTest.java

    
                  private Topology createStatefulTopology(final String storeName) {

                      return topology

                          .addSource("source", STRING_DESERIALIZER, STRING_DESERIALIZER, INPUT_TOPIC_1)

                          .addProcessor("processor", define(new OldAPIStatefulProcessor(storeName)), "source")

                          .addProcessor("processor", () -> new StatefulProcessor(storeName), "source")

Contributor

pegasas Dec 13, 2024

Indeed, it looks like OldAPIStatefulProcessor has been corrected to StatefulProcessor for migrating to current API. Got it! Thanks for teaching!

pegasas approved these changes

View reviewed changes

Member

chia7712 commented Dec 15, 2024

@mjsax could you please fix the conflicts ?


          KAFKA-12829: Remove deprecated Topology#addProcessor of old Processor…

e63d9a3

… API

mjsax force-pushed the kafka-12829-remove-topology-process branch from 0b3c223 to e63d9a3 Compare

December 16, 2024 19:12

mjsax mentioned this pull request

KAFKA-12829: Remove old Processor and ProcessorSupplier interfaces #18238

Merged

chia7712 approved these changes

View reviewed changes

Member

chia7712 left a comment

@mjsax thanks for this patch. LGTM and one small question is left.

streams/test-utils/src/test/java/org/apache/kafka/streams/MockProcessorContextTest.java

-              import static org.mockito.Mockito.when;
-              @SuppressWarnings("deprecation") // this is a test of a deprecated API
-              public class MockProcessorContextTest {

Member

chia7712 Dec 18, 2024

The deprecated org.apache.kafka.streams.processor.MockProcessorContext still exists in the codebase. Should we retain the unit tests?

Member

lucasbru Dec 18, 2024

Good question. I suppose the trouble is that the unit test uses the removed classes, but the unit under test will not be removed yet. Can the tests be rewritten with transformers?

Member Author

mjsax Dec 19, 2024

Good catch.

bbejeck approved these changes

View reviewed changes

Member

bbejeck left a comment

Thanks for the PR @mjsax - LGTM

lucasbru approved these changes

View reviewed changes

Member

lucasbru left a comment

Yes, LGTM modulo the question raised by Chia


          review comments

2423c65

Member Author

mjsax commented Dec 19, 2024

Java 23:

Found 1 test failures:
FAILED ❌ PlaintextAdminIntegrationTest > testConsumerGroupsDeprecatedConsumerGroupState(String, String).quorum=kraft.groupProtocol=consumer
Found 1 flaky test failures:
FLAKY ⚠️  SslTransportLayerTest > "testSelectorPollReadSize(Args).args=tlsProtocol=TLSv1.2, useInlinePem=false"

Java 17:

Found 5 test failures:
FAILED ❌ AuthorizerIntegrationTest > testConsumeUsingAssignWithNoAccess(String, String).quorum=kraft.groupProtocol=consumer
FAILED ❌ PlaintextProducerSendTest > testCloseWithZeroTimeoutFromCallerThread(String, String).quorum=kraft.groupProtocol=classic
FAILED ❌ RemoteLogManagerTest > testRLMOpsWhenMetadataIsNotReady()
FAILED ❌ PlaintextAdminIntegrationTest > testConsumerGroupsDeprecatedConsumerGroupState(String, String).quorum=kraft.groupProtocol=consumer
FAILED ❌ PlaintextAdminIntegrationTest > testConsumerGroups(String, String).quorum=kraft.groupProtocol=consumer
Found 2 flaky test failures:
FLAKY ⚠️  TransactionsWithTieredStoreTest > "testBumpTransactionalEpochWithTV2Enabled(String, String, boolean).quorum=kraft, groupProtocol=consumer, isTV2Enabled=true"
FLAKY ⚠️  SslTransportLayerTest > "testSelectorPollReadSize(Args).args=tlsProtocol=TLSv1.2, useInlinePem=false"

mjsax merged commit 9aa900f into apache:trunk

9 checks passed

mjsax deleted the kafka-12829-remove-topology-process branch

December 20, 2024 00:11

mjsax added a commit that referenced this pull request


          KAFKA-12829: Remove deprecated Topology#addProcessor of old Processor…

be1fe2b

… API (#18154)

Reviewers: John Huang <[email protected]>, Chia-Ping Tsai <[email protected]>, Bill Bejeck <[email protected]>, Lucas Brutschy <[email protected]>

Member Author

mjsax commented Dec 20, 2024

Merged to trunk and cherry-picked to 4.0 branch.

ijuma added a commit to ijuma/kafka that referenced this pull request


          Merge remote-tracking branch 'apache-github/trunk' into kip-896-remov…

31a4619

…e-old-protocol-versions

* apache-github/trunk:
  KAFKA-18312: Added entityType: topicName to SubscribedTopicNames in ShareGroupHeartbeatRequest.json (apache#18285)
  HOTFIX: fix incompatible types: Optional<TimestampAndOffset> cannot be converted to Option<TimestampAndOffset> (apache#18284)
  MINOR Fix some test-catalog issues (apache#18272)
  KAFKA-18180: Move OffsetResultHolder to storage module (apache#18100)
  KAFKA-18301; Make coordinator records first class citizen (apache#18261)
  KAFKA-18262 Remove DefaultPartitioner and UniformStickyPartitioner (apache#18204)
  KAFKA-18296 Remove deprecated KafkaBasedLog constructor (apache#18257)
  KAFKA-12829: Remove old Processor and ProcessorSupplier interfaces (apache#18238)
  KAFKA-18292 Remove deprecated methods of UpdateFeaturesOptions (apache#18245)
  KAFKA-12829: Remove deprecated Topology#addProcessor of old Processor API (apache#18154)
  KAFKA-18035, KAFKA-18306, KAFKA-18092: Address TransactionsTest flaky tests (apache#18264)
  MINOR: change the default linger time in the new coordinator (apache#18274)
  KAFKA-18305: validate controller.listener.names is not in inter.broker.listener.name for kcontrollers (apache#18222)
  KAFKA-18207: Serde for handling transaction records (apache#18136)
  KAFKA-13722: Refactor Kafka Streams store interfaces (apache#18243)
  KAFKA-17131: Refactor TimeDefinitions (apache#18241)
  MINOR: Fix MessageFormatters (apache#18266)
  Mark flaky tests for Dec 18, 2024 (apache#18263)

tedyu pushed a commit to tedyu/kafka that referenced this pull request


          KAFKA-12829: Remove deprecated Topology#addProcessor of old Processor…

6891ee2

… API (apache#18154)

Reviewers: John Huang <[email protected]>, Chia-Ping Tsai <[email protected]>, Bill Bejeck <[email protected]>, Lucas Brutschy <[email protected]>

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels