KAFKA-16339: [1/4 KStream#transform] Remove Deprecated "transformer" methods and classes by fonsdant · Pull Request #17198 · apache/kafka

fonsdant · 2024-09-15T00:44:23Z

Hi, @mjsax! I have removed the transform methods from the main and test classes. However, I have asked myself whether the approach I have used for StandbyTaskEOSIntegrationTest and StreamThreadTest is correct. Should I replace transform with process as suggested in transform Javadoc or remove these tests? Thanks!

fonsdant · 2024-09-16T00:41:22Z

I have missed removing the references to transform on Javadocs too. I will push another commit for it.

fonsdant · 2024-09-17T22:52:56Z

I have noticed that KStreamTransformIntegrationTest.TestTransformer could be removed as it has become unused.

fonsdant · 2024-09-17T23:23:23Z

For now, I have removed the build with deduplicaiton topology test until we decide how to replace the transform method inside it.

fonsdant · 2024-09-20T00:37:42Z

I have replaced tranform with process as suggested by docs. I have implemented Processor based on StandbyTaskEOSMultiRebalanceIntegrationTest.buildWithUniqueIdAssignmentTopology, but shouldWipeOutStandbyStateDirectoryIfCheckpointIsMissing has failed.

fonsdant · 2024-09-20T01:22:19Z

I have managed to fix the test! :D

fonsdant · 2024-09-23T16:15:19Z

I have removed the references to transform from docs. I am wondering if it would be better to replace transform with process, instead of removing it. I have started to think about it while removing the flatTransformValues (the third method to remove). @mjsax, could you help me with this decision?

mjsax

Thanks for the PR. Made a pass.

Btw: is there anything in docs/streams/... that we need to update/remove, too (we can also do this as a single docs-only PR for the whole ticket, ie, all transformXxx() method)

mjsax · 2024-09-28T15:55:34Z

     * Thus, an input record {@code <K,V>} can be transformed into an output record {@code <K':V'>}.
-     * This is a stateless record-by-record operation (cf. {@link #transform(TransformerSupplier, String...)} for
-     * stateful record transformation).
+     * This is a stateless record-by-record operation.


Yes, it think we should link to #process(...) here, instead of removing (same elsewhere where it make sense).

Btw: feel free to skip updating JavaDocs of other deprecated methods which will be remove in follow up PRs of this ticket.

mjsax · 2024-09-28T16:11:35Z

                        }

-                        return KeyValue.pair(key, value);
+                        store.put(key, value);


We do we add a put here?

Similar to the return null replacement with delete. Since we would return a pair with key and value, now I put it in store.

mjsax · 2024-09-28T16:15:08Z

-    @ParameterizedTest
-    @MethodSource("data")        
-    @SuppressWarnings("deprecation")
-    public void shouldPunctuateWithTimestampPreservedInProcessorContext(final boolean stateUpdaterEnabled, final boolean processingThreadsEnabled) {


I think we need to keep this test, but rewrite using #process().

mjsax · 2024-09-28T16:16:49Z


-  @nowarn
-  @Test
-  def shouldBuildIdenticalTopologyInJavaNScalaTransform(): Unit = {


Wondering if we have a test gap, as we don't have a similar test for process() -- should we add a new test for #process() like this one (or rewrite this test for #process())?

I have rewritten this test with process, but have missed to rewrite the test name. I will fix it.

mjsax · 2024-09-28T16:18:02Z


-  @nowarn
-  @Test
-  def testTransformCorrectlyRecords(): Unit = {


Same question as above. Seems we are lacking a similar test for #process() ?

Hm... I have replaced Transformer with ProcessorSupplier and transform with process. Have I missed something else?

mjsax · 2024-09-28T16:18:20Z

-  @nowarn
-  @Test
-  def testSettingNameOnTransform(): Unit = {
-    class TestTransformer extends Transformer[String, String, KeyValue[String, String]] {


fonsdant · 2024-10-08T11:38:59Z

I will push some commits, one for each file from which I have removed the transform link, replacing it with process.

fonsdant · 2024-10-14T21:14:21Z

I have rebased my branch on trunk and this caused GitHub to close the PR. I am implementing the last review points again.

fonsdant · 2024-10-15T12:21:32Z

This time, I have focused on removing the methods first and doing refactorings. After this, I will proceed with Javadoc updates.

I have made the removing part. Now, I will do the refactoring part.

mjsax

Thanks for updating the PR. LGTM.

Just triggered CI. Can merge after it finished and passed.

mjsax · 2024-10-18T22:48:03Z

@fonsdant -- seems there is some compilation error:

StandbyTaskEOSIntegrationTest.java:332: error: cannot find symbol

            .transform(
> Task :streams:compileTestJava
            ^
  symbol:   method transform(()->new or[...]{ } },String)
  location: interface KStream<Integer,Integer>

Seem we need to rewrite this test to use the new process() instead of transform()

fonsdant · 2024-10-19T19:29:25Z

Thanks, @mjsax, still need to refactor it. Meanwhile, do you prefer that I convert this PR to draft? Or should I keep it as it is?

mjsax · 2024-10-22T00:29:38Z

We can keep this PR as-is -- you can also to the refactoring as part of this PR if you want.

mjsax · 2024-10-30T06:12:27Z

@fonsdant Any update on this PR? -- We are approaching AK 4.0 release deadlines, and there is 3 more PRs for this ticket...

fonsdant · 2024-10-30T10:19:56Z

@mjsax, sorry for the delay, I am still refactoring the tests, replacing transform with process.

Given the coming deadline, I agree it would be adequate someone more experienced to resume it.

fonsdant · 2024-10-30T23:27:17Z

I will push the work I done this passed week.

mjsax · 2024-10-31T00:45:16Z

Given the coming deadline, I agree it would be adequate someone more experienced to resume it.

As long as you have enough time to keep working, I think we are a good. Of course, if you are blocked, let us know.

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

mjsax · 2024-11-04T21:00:24Z

+//
+//    val transformNode = builder.build().describe().subtopologies().asScala.head.nodes().asScala.toList(1)
+//    assertEquals("my-name", transformNode.name())
+//  }


Should we rewrite this to testSettingNameOnProcess?

Oh, sorry! I have missed it up. Of course, I will rewrite it!

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

fonsdant · 2024-11-05T01:31:04Z

@mjsax, I have removed the adding lines in buildTopologyWithDeduplication in StandbyTaskEOSIntegrationTest to match the previous implementation with transform (but still with process). Also, removed the comments of the KStreamTest scala test and rewritten it. Finally, removed var usage.

mjsax

Thanks for the quick turn around. I think the Scala test can be simplified significantly. In the end, we only verify the name of he processor node in the Topology.

mjsax · 2024-11-05T02:06:59Z

+  @Test
+  def testSettingNameOnTransform(): Unit = {
+    val processorSupplier: ProcessorSupplier[String, String, String, String] =
+      new api.ProcessorSupplier[String, String, String, String] {


Why do we need an explicit ProcessorSupplier for this test? And we use process(() => new TestProcessor) similar to the old code?

Oh, it is true... I have made this way, because IntelliJ highlight the "TestProcessor" as an error --- probably conflicting with the deprecated process. See:

But I can actually run the test this way:

Not sure why IntelliJ shows an error... given that TestProcessor is api.Processor I would expect IntelliJ to figure out that the lambda is api.ProcessorSupplier and resolve to the new process(...) overload... In any case, the build did pass, so it seem ok. Also, the old process() will be removed soon, too: #17190

mjsax · 2024-11-05T02:07:19Z

+  def testSettingNameOnTransform(): Unit = {
+    val processorSupplier: ProcessorSupplier[String, String, String, String] =
+      new api.ProcessorSupplier[String, String, String, String] {
+        private val storeName = "store-name"


Why do we add a store?

mjsax · 2024-11-05T02:08:40Z

+              val processedKey = s"$key-processed"
+              val processedValue = s"$value-processed"
+              store.put(processedKey, processedValue)
+              context.forward(new api.Record(processedKey, processedValue, record.timestamp()))


What the Processor does is actually pretty meaningless for the test, so I think the while implementation could actually be empty? For this case, we don't even need init() method, as we don't need access to the context either.

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

fonsdant · 2024-11-05T10:22:34Z

Thanks, @mjsax! I have simplified the test and replaced the processorSupplier with a lambda. Should I replace it for for other methods too? And how about IntelliJ reporting it as problem?

mjsax · 2024-11-06T06:05:37Z

Merged to trunk. Thanks for the PR! One down. Three to go. :) Looking forward to your update of #17245

fonsdant · 2024-11-06T09:35:42Z

Uhuu! Here we go! I need to rebase the second one on this first now. Thanks, @mjsax!

…methods and classes (apache#17198) Reviewers: Matthias J. Sax <matthias@confluent.io>

fonsdant changed the title ~~KAFKA-16339: [1/4 KStream#tranform] Remove Deprecated "transformer" methods and classes~~ KAFKA-16339: [1/4 KStream#transform] Remove Deprecated "transformer" methods and classes Sep 15, 2024

fonsdant force-pushed the kstream-transform-remove-deprecated-transformer-methods-and-classes branch 2 times, most recently from bafaa10 to f28ac72 Compare September 17, 2024 23:17

github-actions Bot added the streams label Sep 20, 2024

fonsdant mentioned this pull request Sep 20, 2024

KAFKA-16339: [2/4 KStream#flatTransform] Remove Deprecated "transformer" methods and classes #17245

Merged

mjsax reviewed Sep 28, 2024

View reviewed changes

kamalcph mentioned this pull request Oct 12, 2024

KAFKA-15859: Make RemoteListOffsets call an async operation #16602

Merged

3 tasks

fonsdant closed this Oct 14, 2024

fonsdant force-pushed the kstream-transform-remove-deprecated-transformer-methods-and-classes branch from 35fd141 to 203f323 Compare October 14, 2024 20:48

github-actions Bot added the small Small PRs label Oct 14, 2024

fonsdant reopened this Oct 14, 2024

github-actions Bot removed the small Small PRs label Oct 14, 2024

mjsax added the ci-approved label Oct 16, 2024

mjsax approved these changes Oct 16, 2024

View reviewed changes

fonsdant added 10 commits November 4, 2024 08:58

refactor scala topology test

2e2ec57

refactor scala kstream test

b090991

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

adjust topology test for spotless scala

46cf8d1

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

adjust kstream test for spotless scala

05d3d7c

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

update javadoc of transformer

5db97e2

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

update javadoc of transformer supplier

42f8635

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

update javadoc of value transformer supplier

38cc9c3

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

update javadoc of value transformer with key supplier

7b51fd1

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

update javadoc of connected store provider

532f9f2

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

update javadoc of kstream

7a65aed

Signed-off-by: Joao Pedro Fonseca Dantas <fonsdant@gmail.com>

fonsdant force-pushed the kstream-transform-remove-deprecated-transformer-methods-and-classes branch from 46a01a6 to 7a65aed Compare November 4, 2024 11:59