TaskCallbackImpl implements TaskCallback, Comparable<TaskCallbackImpl> { @Override public void complete() { if (scheduledFuture != null) { scheduledFuture.cancel(true); } log.trace("Callback complete for task {}, ssp {}, offset {}.", new Object[] {taskName, systemStreamPartition, offset}); if (isComplete.compareAndSet(false, true)) { listener.onComplete(this); } else { String msg = String.format("Callback complete was invoked after completion for task %s, ssp %s, offset %s.", taskName, systemStreamPartition, offset); listener.onFailure(this, new IllegalStateException(msg)); } } 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 testComplete() { callback.complete(); assertEquals(1L, completeCount.get()); assertEquals(0L, failureCount.get()); }
@Test public void testCallbackMultipleComplete() { callback.complete(); assertEquals(1L, completeCount.get()); callback.complete(); assertEquals(1L, failureCount.get()); assertTrue(throwable instanceof IllegalStateException); }
@Test public void testMultithreadedCallbacks() throws Exception { final CyclicBarrier barrier = new CyclicBarrier(2); ExecutorService executor = Executors.newFixedThreadPool(2); for (int i = 0; i < 2; i++) { executor.submit(new Runnable() { @Override public void run() { try { barrier.await(); callback.complete(); } catch (Exception e) { e.printStackTrace(); } } }); } executor.awaitTermination(1, TimeUnit.SECONDS); assertEquals(1L, completeCount.get()); assertEquals(1L, failureCount.get()); assertTrue(throwable instanceof IllegalStateException); }