@Override
        public TaskCallback createCallback() {
          state.startProcess();
          containerMetrics.processes().inc();
          // report 1 whenever the contaienr is running. Can be used to calculate the number of containers not running
          containerMetrics.containerRunning().set(1L);

          /*
           * Timeout used in the task callback. The value is determined based on the following logic
           * 1. If run loop is in draining mode and the envelope is drain, use drainCallbackTimeoutMs
           * 2. If the envelope is watermark, use watermarkCallbackTimeoutMs regardless of the modes. Setting a lower
           *    watermark callback timeout during draining mode can cause drain to be unsuccessful prematurely and
           *    vice-versa.
           * 3. Use callbackTimeoutMs otherwise
           */
          long timeout = messageCallbackTimeoutMs;

          if (envelope.isWatermark()) {
            timeout = watermarkCallbackTimeoutMs;
          } else if (isDraining && envelope.isDrain()) {
            timeout = drainCallbackTimeoutMs;
          }

          return callbackManager.createCallback(task.taskName(), envelope, coordinator, timeout);
        }
Process asynchronously. The callback needs to be fired once the processing is done.
samza-core/src/main/java/org/apache/samza/container/RunLoop.java