@Override
  public ExecutorService getOperatorExecutor(TaskName taskName, Config config) {
    ExecutorService taskExecutor = TASK_EXECUTORS.computeIfAbsent(taskName, key -> {
      final int threadPoolSize = new JobConfig(config).getThreadPoolSize();
      ExecutorService operatorExecutor;

      if (threadPoolSize > 1) {
        LOG.info("Using container thread pool as operator thread pool for task {}", key.getTaskName());
        operatorExecutor = getTaskExecutor(config);
      } else {
        LOG.info("Using single threaded thread pool as operator thread pool for task {}", key.getTaskName());
        operatorExecutor = Executors.newSingleThreadExecutor(
            new ThreadFactoryBuilder().setNameFormat("Samza " + key.getTaskName() + " Thread-%d")
                .setDaemon(true)
                .build());
      }

      return operatorExecutor;
    });

    return taskExecutor;
  }
{@inheritDoc}

The choice of thread pool is determined based on the following logic
    1. Use {@link #getTaskExecutor(Config)} if job.container.thread.pool.size &gt; 1
    2. 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.
samza-core/src/main/java/org/apache/samza/task/DefaultTaskExecutorFactory.java