lucasbru commented on code in PR #14193:
URL: https://github.com/apache/kafka/pull/14193#discussion_r1348755156
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -452,6 +468,31 @@ public static StreamThread create(final TopologyMetadata
topologyMetadata,
return
streamThread.updateThreadMetadata(getSharedAdminClientId(clientId));
}
+ private static DefaultTaskManager maybeCreateSchedulingTaskManager(final
boolean processingThreadsEnabled,
+ final
boolean stateUpdaterEnabled,
+ final
TopologyMetadata topologyMetadata,
+ final
Time time,
+ final
String threadId,
+ final
Tasks tasks) {
+ if (processingThreadsEnabled) {
+ if (!stateUpdaterEnabled) {
+ throw new IllegalStateException("Processing threads require
the state updater to be enabled");
+ }
+
+ final DefaultTaskManager defaultTaskManager = new
DefaultTaskManager(
+ time,
+ threadId,
Review Comment:
Yes exactly. I would keep this name for now. Actually, we may even want to
keep the ability to run multiple polling threads, just not do it by default.
But either way, at the moment there will be multiple StreamThreads, each with
its own task executors.
--
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
