TaskCallbackImpl implements TaskCallback, Comparable<TaskCallbackImpl> { @Override public void failure(Throwable t) { if (scheduledFuture != null) { scheduledFuture.cancel(true); } if (isComplete.compareAndSet(false, true)) { String msg = String.format("Callback failed for task %s, ssp %s, offset %s.", taskName, systemStreamPartition, offset); listener.onFailure(this, new SamzaException(msg, t)); } else { String msg = String.format("Task callback failure was invoked after completion for task %s, ssp %s, offset %s.", taskName, systemStreamPartition, offset); listener.onFailure(this, new IllegalStateException(msg, t)); } } TaskCallbackImpl(TaskCallbackListener listener,
      TaskName taskName,
      IncomingMessageEnvelope envelope,
      ReadableCoordinator coordinator,
      long seqNum,
      long timeCreatedNs); TaskName getTaskName(); String getOffset(); SystemStreamPartition getSystemStreamPartition(); ReadableCoordinator getCoordinator(); long getTimeCreatedNs(); @Override void complete(); @Override void failure(Throwable t); @Override int compareTo(TaskCallbackImpl callback);  }
@Test public void testFailure() { callback.failure(new Exception("dummy exception")); assertEquals(0L, completeCount.get()); assertEquals(1L, failureCount.get()); }