mynameborat commented on code in PR #1667:
URL: https://github.com/apache/samza/pull/1667#discussion_r1227325284
##########
samza-core/src/main/java/org/apache/samza/task/DefaultTaskExecutorFactory.java:
##########
@@ -37,4 +45,36 @@ public ExecutorService getTaskExecutor(Config config) {
return Executors.newFixedThreadPool(threadPoolSize,
new ThreadFactoryBuilder().setNameFormat("Samza Container
Thread-%d").build());
}
+
+ /**
+ * {@inheritDoc}
+ *
+ * The choice of thread pool is determined based on the following logic
+ * If job.operator.thread.pool.enabled,
+ * a. Use {@link #getTaskExecutor(Config)} if
job.container.thread.pool.size > 1
+ * b. Use default single threaded pool otherwise
+ * <b>Note:</b> The default single threaded pool used is a substitute for
the scenario where container thread pool is null and
+ * the messages are dispatched on runloop thread. We can't have the stages
schedule on the run loop thread and hence
+ * the fallback to use a single threaded executor across all tasks.
+ */
+ @Override
+ public ExecutorService getOperatorExecutor(TaskName taskName, Config config)
{
+ ExecutorService taskExecutor = TASK_EXECUTORS.computeIfAbsent(taskName,
key -> {
+ final int threadPoolSize = new JobConfig(config).getThreadPoolSize();
+ ExecutorService operatorThreadPool;
+
+ if (threadPoolSize > 1) {
+ LOG.info("Using container thread pool as operator thread pool for task
{}", key.getTaskName());
+ operatorThreadPool = getTaskExecutor(config);
Review Comment:
Note: Other reasons to fallaback to container thread pool is so that we can
tune the thread using existing known configurations and introducing a new
configuration and tunable may need to be thought through in order to prevent
proliferation of configs, tunables combinations.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
