TaskCallbackManager { public TaskCallbackImpl createCallback(TaskName taskName, IncomingMessageEnvelope envelope, ReadableCoordinator coordinator) { 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 " + timeout + " ms."; callback.failure(new SamzaException(msg)); } }; ScheduledFuture scheduledFuture = timer.schedule(timerTask, timeout, TimeUnit.MILLISECONDS); callback.setScheduledFuture(scheduledFuture); } return callback; } TaskCallbackManager(TaskCallbackListener listener,
      ScheduledExecutorService timer,
      long timeout,
      int maxConcurrency,
      HighResolutionClock clock); TaskCallbackImpl createCallback(TaskName taskName,
      IncomingMessageEnvelope envelope,
      ReadableCoordinator coordinator); List<TaskCallbackImpl> updateCallback(TaskCallbackImpl callback);  }
@Test public void testCreateCallback() { TaskCallbackImpl callback = callbackManager.createCallback(new TaskName("Partition 0"), mock(IncomingMessageEnvelope.class), null); assertTrue(callback.matchSeqNum(0)); callback = callbackManager.createCallback(new TaskName("Partition 0"), mock(IncomingMessageEnvelope.class), null); assertTrue(callback.matchSeqNum(1)); }