KAFKA-19942: Clean up StreamThread and StoreChangelogReader Test #21022
Conversation
github-actions
bot
added
triage
|
@lucasbru, tagging for review :) |
| @@ -1396,7 +1363,7 @@ public void shouldOnlyCompleteShutdownAfterRebalanceNotInProgress(final boolean | |||
| assertTrue(thread.isAlive()); | |||
|
|
|||
| Thread.sleep(1000); | |||
| assertEquals(Set.of(task1, task2), thread.taskManager().activeTaskIds()); | |||
| assertEquals(Set.of(task1, task2), thread.taskManager().allTasks().keySet()); | |||
There was a problem hiding this comment.
We use allTasks() here because the tasks, at this moment, may also be in the initialization queue and not active.
tasks.addPendingTasksToInit(newActiveTasks)
shashankhs11
changed the title
There was a problem hiding this comment.
Pull request overview
This PR removes the old state updater toggle infrastructure as part of KAFKA-18913 cleanup work. The state updater is now always enabled, simplifying the codebase by removing conditional logic and duplicate test methods.
- Removed
stateUpdaterEnabledparameter from all test methods - Converted parameterized tests from
@MethodSource("data")to@ValueSource(booleans = {true, false})forprocessingThreadsEnabled - Hardcoded state updater configuration to always be enabled
Reviewed changes
Copilot reviewed 2 out of 3 changed files in this pull request and generated 4 comments.
| File | Description |
|---|---|
| StreamThreadTest.java | Removed stateUpdaterEnabled parameter from all tests, deleted unused imports and helper methods, simplified test logic by removing state updater conditionals |
| StoreChangelogReaderTest.java | Consolidated duplicate test methods, removed state updater parameter and conditional logic |
| StreamsConfig.java | Modified InternalConfig.stateUpdaterEnabled() to always return true |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| }; | ||
| } | ||
| throw taskMigratedException; | ||
| }); |
There was a problem hiding this comment.
Trailing whitespace detected. Please remove the trailing whitespace on this line for consistency with the project's code style.
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
Outdated
Show resolved
Hide resolved
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
Outdated
Show resolved
Hide resolved
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java
Outdated
Show resolved
Hide resolved
|
|
||
| public static boolean stateUpdaterEnabled(final Map<String, Object> configs) { | ||
| return InternalConfig.getBoolean(configs, InternalConfig.STATE_UPDATER_ENABLED, true); | ||
| return true; // always enabled |
There was a problem hiding this comment.
I suppose this is temporary and will be removed soon
There was a problem hiding this comment.
Yes! The config will be removed in a follow-up PR.
I have all the changes ready and will open a PR as
soon as we merge this one.
…nals/StreamThreadTest.java Co-authored-by: Copilot <[email protected]>
…nals/StreamThreadTest.java Co-authored-by: Copilot <[email protected]>
…nals/StreamThreadTest.java Co-authored-by: Copilot <[email protected]>
2 participants
This PR is continuation of the work related to cleaning up of the old
stateupdater code and is a sub-task of
KAFKA-18913
StreamThreadTest.javaStoreChangelogReaderTest.javaReviewers: Lucas Brutschy [email protected]