public TaskCallbackImpl createCallback(TaskName taskName,
      IncomingMessageEnvelope envelope,
      ReadableCoordinator coordinator,
      long callbackTimeout) {
    final TaskCallbackImpl callback =
        new TaskCallbackImpl(listener, taskName, envelope, coordinator, seqNum++, clock.nanoTime());
    if (timer != null) {
      Runnable timerTask = new Runnable() {
        @Override
        public void run() {
          ThreadUtil.logThreadDump("Thread dump at task callback timeout");
          String msg = "Callback for task {} " + callback.taskName + " timed out after " + callbackTimeout + " ms.";
          callback.failure(new SamzaException(msg));
        }
      };

      final ScheduledFuture scheduledFuture = timer.schedule(timerTask, callbackTimeout, TimeUnit.MILLISECONDS);
      callback.setScheduledFuture(scheduledFuture);
    }

    return callback;
  }
Creates a task callback.
@param taskName task name
@param envelope incoming envelope
@param coordinator coordinator
@param callbackTimeout timeout to expire the callback
samza-core/src/main/java/org/apache/samza/task/TaskCallbackManager.java