KAFKA-12829: Remove the deprecated method `init(ProcessorContext, StateStore)` from the `StateStore` interface by xijiu · Pull Request #16906 · apache/kafka

xijiu · 2024-08-17T09:15:55Z

See the discussion: (#11611 (comment))

I did the following three things:

Find all the code that has invoked the init (ProcessorContext, StateStore) method. Actually, it is only called in one test class(MockProcessorContextTest), then I deleted the relevant test method.
Delete the method init(ProcessorContext, StateStore) from the StateStore and Its implementation classes. If the implementation class contains both methods init(ProcessorContext, StateStore) and init(StateStoreContext, StateStore) simultaneously, then delete init(ProcessorContext, StateStore) directly. If the implementation class only contains methods init(ProcessorContext, StateStore), then I create a new method init(StateStoreContext, StateStore) and copy the relevant code to it. There are still many details in the code.
The logic of generation of internal topic names in class TimeOrderedCachingWindowStore was not repalaced by ProcessorContextUtils#changelogFor, I moved it over while I was at it.

Committer Checklist (excluded from commit message)

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

xijiu · 2024-08-17T09:16:22Z

After that, I ran all the junit test in the module streams, and all of the tests are passed in about 30 min.

chia7712

@xijiu thanks for your patch. one major comment is left.

chia7712 · 2024-08-20T02:04:46Z

-    default void init(final StateStoreContext context, final StateStore root) {
-        init(StoreToProcessorContextAdapter.adapt(context), root);
-    }
+    default void init(final StateStoreContext context, final StateStore root) {}


This is used to replace the deprecated init, so I guess it should NOT have default implementation after removing the deprecated version?

@guozhangwang WDYT?

chia7712

@xijiu thanks for your patch. Please remove all initInternal to simplify code.

chia7712 · 2024-08-22T01:38:47Z

-    @Override
-    public void init(final ProcessorContext context,
-                     final StateStore root) {
+    private void initInternal(final ProcessorContext context,


Could you please merge initInternal into init? we don't need to have a private method used to generate a small public method :)

chia7712 · 2024-08-22T01:38:53Z

-    @Override
-    public void init(final ProcessorContext context,
-                     final StateStore root) {
+    private void initInternal(final ProcessorContext context,


xijiu · 2024-08-22T15:27:17Z

@chia7712 Thanks for the code review, I have fixed it, PTAL again

mjsax · 2024-08-22T19:53:57Z

-        this.context = context;
+    public void init(final StateStoreContext stateStoreContext, final StateStore root) {
+        this.stateStoreContext = stateStoreContext;
+        this.context = StoreToProcessorContextAdapter.adapt(stateStoreContext);


This feels rather hacky -- using adapt was a workaround as long as we have both init() methods. I think, we need to do more cleanup/refactoring and get rid or this.context completely.

In other stores, we use InternalProcessorContext -- I think doing the same here would be the right way forward (and get rid of both stateStoreContext and context in favor of internalProcessorContext)?

Applies elsewhere, too.

(Ideally, we should be able to delete class StoreToProcessorContextAdapter entirely)

mjsax · 2024-08-22T20:00:52Z

-    public void init(final ProcessorContext context, final StateStore root) {
-        throw new UnsupportedOperationException("cannot initialize a logical segment");
-    }
+    public void init(final StateStoreContext context, final StateStore root) {}


This method should throw (as the old one)

mjsax · 2024-08-22T20:02:55Z

-    @Override
-    public void init(final ProcessorContext context,
-                     final StateStore root) {
-        initInternal(asInternalProcessorContext(context));


If we call initInternal only in a single place now, we might consider inlining it?

agree. @xijiu please apply "inline" to elsewhere, too

mjsax · 2024-08-22T20:09:52Z

-    public void init(final ProcessorContext context,
+    public void init(final StateStoreContext stateStoreContext,
                     final StateStore root) {
+        final ProcessorContext context = StoreToProcessorContextAdapter.adapt(stateStoreContext);


Do we need context at all? Seem we can use stateStoreContext below got call appConfigs() and register() ?

mjsax · 2024-08-22T20:13:08Z

-        this.context = context;
-        final String topic = ProcessorStateManager.storeChangelogTopic(prefix, name(),  context.taskId().topologyName());
+        this.context = asInternalProcessorContext(context);
+        final String topic = ProcessorContextUtils.changelogFor(context, name(), Boolean.TRUE);


Can we exclude this change from this PR, as this PR should only remove init() to address KAFKA-12829, and pull this change into a follow up PR for KAFKA-13588.

mjsax · 2024-08-22T20:16:55Z

    }

-    @Test
-    public void shouldStoreAndReturnStateStores() {


Why do we remove this test? Should we rather rewrite it calling the new init() passing in a mocked StateStoreContext object?

mjsax

Thanks for the PR. Made a pass. This is a complex change. Not sure if I caught everything... Might find more in a second pass after the PR was updated.

xijiu · 2024-08-24T05:11:04Z

@mjsax @chia7712 Thanks very much for these professional comments, they have greatly improved my understanding of the work.
I have revised the code, PTAL again

mjsax

Thanks for updating the PR. A few more nits. Also, there are failing tests which seems to be related to the changes of this PR.

mjsax · 2024-08-25T20:53:09Z

-
    @Override
    public void init(final StateStoreContext context, final StateStore root) {
        final String changelogTopic = ProcessorContextUtils.changelogFor(context, name(), Boolean.TRUE);


Can we inline initInternal here, too?

mjsax · 2024-08-25T20:56:38Z

-    public void init(final ProcessorContext context, final StateStore root) {
-        this.context = ProcessorContextUtils.asInternalProcessorContext(context);
-        changelogTopic = ProcessorContextUtils.changelogFor(context, name(), Boolean.TRUE);
-        init(root);


Seem this init is the same and internalInit in other classes -- can we inline it?

xijiu · 2024-08-26T13:48:26Z

hi @mjsax , I have improved the code based on the suggestions, include some junit test. And I also ran the stream moudle
all tests successfully before I update the PR.

PTAL, Thanks

mjsax · 2024-08-26T21:16:17Z

    }
+
+
+    public InternalProcessorContext getInternalProcessorContext() {


MockProcessorContext is public API and we cannot just add this. -- I assume this was necessary to fix some tests? For this case, we should do it differently.

mjsax · 2024-08-26T21:16:25Z


    <suppress checks="MethodLength"
-              files="KTableImpl.java"/>
+              files="(KTableImpl|MockProcessorContext).java"/>


I hope we don't need this

mjsax · 2024-08-26T21:26:43Z

-            this.context = null;
-            expiredRecordSensor = null;
-        }
+        this.context = asInternalProcessorContext(stateStoreContext);


It seems this change breaks the test -- let's not apply this change but keep the code as-is.

…cessorContext

xijiu · 2024-08-28T16:49:20Z

hello @mjsax , I have fixed the code, PTAL

mjsax · 2024-08-29T00:49:46Z

Thanks a ton for this PR @xijiu! Huge piece of work! Great job!

Merged to trunk.

xijiu · 2024-08-29T03:18:38Z

Thanks a ton for this PR @xijiu! Huge piece of work! Great job!

Merged to trunk.

@mjsax I learned a lot during this process and gained a deeper understanding of streams. Thank you very much for code review and guidance.

…teStore)` from the `StateStore` interface (apache#16906) Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, Matthias J. Sax <matthias@confluent.io>

chia7712 reviewed Aug 20, 2024

View reviewed changes

chia7712 reviewed Aug 22, 2024

View reviewed changes

mjsax added the streams label Aug 22, 2024

mjsax requested a review from guozhangwang August 22, 2024 03:45

mjsax changed the title ~~KAFKA-13588: Remove the deprecated method init(ProcessorContext, StateStore) from the StateStore interface~~ KAFKA-12829: Remove the deprecated method init(ProcessorContext, StateStore) from the StateStore interface Aug 22, 2024

mjsax reviewed Aug 22, 2024

View reviewed changes

mjsax mentioned this pull request Aug 23, 2024

KAFKA-12833 Remove Deprecated methods under TopologyTestDriver #16804

Closed

3 tasks

xijiu force-pushed the remove_deprecated_init_method branch from 644f98c to 512acb8 Compare August 24, 2024 04:49

mjsax reviewed Aug 25, 2024

View reviewed changes

xijiu force-pushed the remove_deprecated_init_method branch from 512acb8 to 4f613c7 Compare August 26, 2024 13:37

mjsax reviewed Aug 26, 2024

View reviewed changes

xijiu added 7 commits August 29, 2024 00:44

clear the init implement code

353c6a4

remove init(ProcessorContext,StateStore) method

74daaa4

code optimization

4d35402

get rid of both stateStoreContext and context in favor of internalPro…

1f65487

…cessorContext

adjust test in MockProcessorContextTest class

7ca0e77

Keep optimizing

f5c08f6

Keep optimizing

0dcddcf

xijiu force-pushed the remove_deprecated_init_method branch from 4f613c7 to 0dcddcf Compare August 28, 2024 16:44

mjsax merged commit 291523e into apache:trunk Aug 29, 2024

This was referenced Mar 8, 2026

KAFKA-19983: Fix MockProcessorContext doesn't work with WindowStores #21675

Closed

KAFKA-19983: Fix MockProcessorContext doesn't work with WindowStores. #21677

Draft

		}


		public InternalProcessorContext getInternalProcessorContext() {

Conversation

xijiu commented Aug 17, 2024

Committer Checklist (excluded from commit message)

Uh oh!

xijiu commented Aug 17, 2024

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 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!

xijiu commented Aug 22, 2024

Uh oh!

mjsax Aug 22, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

xijiu commented Aug 24, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

xijiu commented Aug 26, 2024

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!

xijiu commented Aug 28, 2024

Uh oh!

mjsax commented Aug 29, 2024

Uh oh!

xijiu commented Aug 29, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mjsax Aug 22, 2024 •

edited

Loading

xijiu commented Aug 24, 2024 •

edited

Loading