KAFKA-18913: Start state updater in task manager

[janchilling]

Updated the code to start the State Updater Thread only after the Stream
Thread is started.

Changes done :

1. Moved the starting of the StateUpdater thread to a new init method in
   the TaskManager.
2. Called the init of TaskManager in the run method of the StreamThread.
3. Updated the test cases in the StreamThreadTest to mimic the
   aforementioned behaviour.

Reviewers: Bruno Cadonna cadonna@apache.org

[janchilling]

Hi @cadonna ,

Here is the fix for the issue that we discussed(StateUpdater thread being left to run even though an IllegalStateException thrown before the StreamThread is created).
I hope this is the implementation that you requested to be done. If there are any changes or if anything needs to be added please let me know. I might be away for the next 2 weeks, I'll attend to any required changes when I find time.

[cadonna]

@janchilling Thanks for the PR!

The PR corresponds to my proposal in the other PR.

I left some comments.

[cadonna]

Could you please add a unit test that verifies that the state updater is indeed started when the task manager is initialized?

[cadonna]

It should be enough to call thread.taskManager().init() instead of this method. Whether the state updater is enabled is already checked in the init().

You also do not need to call it in each and every test method. Only in some the call is needed.

[janchilling]

Hi @cadonna ,

I actually didn't call it in every test method, only in the ones where the StreamThread is created by using the StreamThreadTest#createStreamThread. Therefore, to keep the behavior uniform with the code, I included it in those test cases.

That said, if the above approach isn’t ideal, another option could be to include the call in the tests where tasks are not Restoring properly. I tried this alternative as well, and all test cases passed with this approach as well.

Happy to adjust it either way, depending on what you think would be best.

[cadonna]

Fair enough!
I checked a couple of test methods and found that in at least two cases maybeRunStateUpdater() was not needed. The two cases are: shouldRecordCommitLatency() and shouldChangeStateInRebalanceListener(). There might still be others.

[cadonna]

I might be away for the next 2 weeks, I'll attend to any required changes when I find time.

No worries! Thanks for letting us know!

[janchilling]

Hi @cadonna ,

I’ve added a test case that verifies both the starting and shutdown behavior of the StateUpdater thread—since we shouldn't leave the thread running after the test.

To perform the verification, I used DefaultStateUpdater directly in order to access the isRunning() method, as the existing mock StateUpdater doesn't expose it. This meant writing a more customized test rather than reusing the existing setup methods.

I believe this addresses all the requested changes, but please let me know if there’s anything else you'd like me to update.

[cadonna]

@janchilling Thanks for the update!

Here my comment.

Almost there.

[cadonna]

You do not need to check for isRunning() here. isRunning() is an implementation detail of the DefaultStateUpdater. It suffices to verify that stateUpdater.start() is called on the mock. Something simple like the following:

    @ParameterizedTest
    @ValueSource(booleans = {true, false})
    public void shouldStartStateUpdaterOnInit(final boolean stateUpdaterEnabled) {
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, stateUpdaterEnabled);
        taskManager.init();
        if (stateUpdaterEnabled) {
            verify(stateUpdater).start();
        } else {
            verify(stateUpdater, never()).start();
        }
    }

The shutdown is tested elsewhere. You do not need to test it again here.

[janchilling]

Thanks for pointing that out. I've updated the test accordingly.

[cadonna]

Thanks @janchilling!

LGTM!