cadonna commented on code in PR #19889:
URL: https://github.com/apache/kafka/pull/19889#discussion_r2125829251
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java:
##########
@@ -149,6 +149,11 @@ public class TaskManager {
);
}
+ void init() {
+ if (stateUpdater != null) {
+ this.stateUpdater.start();
+ }
+ }
Review Comment:
Could you please add a unit test that verifies that the state updater is
indeed started when the task manager is initialized?
##########
streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java:
##########
@@ -4117,6 +4149,12 @@ private StreamThread buildStreamThread(final
Consumer<byte[], byte[]> consumer,
null
);
}
+
+ private void maybeRunStateUpdater(final boolean stateUpdaterEnabled) {
+ if (stateUpdaterEnabled) {
+ thread.taskManager().init();
+ }
+ }
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
