@Override
  public void processAsync(final IncomingMessageEnvelope envelope,
      final MessageCollector collector,
      final TaskCoordinator coordinator,
      final TaskCallback callback) {
    if (executor != null) {
      executor.submit(new Runnable() {
        @Override
        public void run() {
          process(envelope, collector, coordinator, callback);
        }
      });
    } else {
      // legacy mode: running all tasks in the runloop thread
      process(envelope, collector, coordinator, callback);
    }
  }
AsyncStreamTaskAdapter allows a StreamTask to be executed in parallel.The class
uses the build-in thread pool to invoke StreamTask.process and triggers
the callbacks once it's done. If the thread pool is null, it follows the legacy
synchronous model to execute the tasks on the run loop thread.
samza-core/src/main/java/org/apache/samza/task/AsyncStreamTaskAdapter.java