TaskCallbackManager { public List<TaskCallbackImpl> updateCallback(TaskCallbackImpl callback) { if (maxConcurrency > 1) { return completedCallbacks.update(callback); } else { return ImmutableList.of(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 testUpdateCallbackInOrder() { TaskName taskName = new TaskName("Partition 0"); SystemStreamPartition ssp = new SystemStreamPartition("kafka", "topic", new Partition(0)); ReadableCoordinator coordinator = new ReadableCoordinator(taskName); IncomingMessageEnvelope envelope0 = new IncomingMessageEnvelope(ssp, "0", null, null); TaskCallbackImpl callback0 = new TaskCallbackImpl(listener, taskName, envelope0, coordinator, 0, 0); List<TaskCallbackImpl> callbacksToUpdate = callbackManager.updateCallback(callback0); assertEquals(1, callbacksToUpdate.size()); TaskCallbackImpl callback = callbacksToUpdate.get(0); assertTrue(callback.matchSeqNum(0)); assertEquals(ssp, callback.getSystemStreamPartition()); assertEquals("0", callback.getOffset()); IncomingMessageEnvelope envelope1 = new IncomingMessageEnvelope(ssp, "1", null, null); TaskCallbackImpl callback1 = new TaskCallbackImpl(listener, taskName, envelope1, coordinator, 1, 0); callbacksToUpdate = callbackManager.updateCallback(callback1); assertEquals(1, callbacksToUpdate.size()); callback = callbacksToUpdate.get(0); assertTrue(callback.matchSeqNum(1)); assertEquals(ssp, callback.getSystemStreamPartition()); assertEquals("1", callback.getOffset()); }
@Test public void testUpdateCallbackOutofOrder() { TaskName taskName = new TaskName("Partition 0"); SystemStreamPartition ssp = new SystemStreamPartition("kafka", "topic", new Partition(0)); ReadableCoordinator coordinator = new ReadableCoordinator(taskName); IncomingMessageEnvelope envelope2 = new IncomingMessageEnvelope(ssp, "2", null, null); TaskCallbackImpl callback2 = new TaskCallbackImpl(listener, taskName, envelope2, coordinator, 2, 0); List<TaskCallbackImpl> callbacksToUpdate = callbackManager.updateCallback(callback2); assertTrue(callbacksToUpdate.isEmpty()); IncomingMessageEnvelope envelope1 = new IncomingMessageEnvelope(ssp, "1", null, null); TaskCallbackImpl callback1 = new TaskCallbackImpl(listener, taskName, envelope1, coordinator, 1, 0); callbacksToUpdate = callbackManager.updateCallback(callback1); assertTrue(callbacksToUpdate.isEmpty()); IncomingMessageEnvelope envelope0 = new IncomingMessageEnvelope(ssp, "0", null, null); TaskCallbackImpl callback0 = new TaskCallbackImpl(listener, taskName, envelope0, coordinator, 0, 0); callbacksToUpdate = callbackManager.updateCallback(callback0); assertEquals(3, callbacksToUpdate.size()); TaskCallbackImpl callback = callbacksToUpdate.get(0); assertTrue(callback.matchSeqNum(0)); assertEquals(ssp, callback.getSystemStreamPartition()); assertEquals("0", callback.getOffset()); callback = callbacksToUpdate.get(1); assertTrue(callback.matchSeqNum(1)); assertEquals(ssp, callback.getSystemStreamPartition()); assertEquals("1", callback.getOffset()); callback = callbacksToUpdate.get(2); assertTrue(callback.matchSeqNum(2)); assertEquals(ssp, callback.getSystemStreamPartition()); assertEquals("2", callback.getOffset()); }
@Test public void testUpdateCallbackWithCoordinatorRequests() { TaskName taskName = new TaskName("Partition 0"); SystemStreamPartition ssp = new SystemStreamPartition("kafka", "topic", new Partition(0)); IncomingMessageEnvelope envelope2 = new IncomingMessageEnvelope(ssp, "2", null, null); ReadableCoordinator coordinator2 = new ReadableCoordinator(taskName); coordinator2.shutdown(TaskCoordinator.RequestScope.ALL_TASKS_IN_CONTAINER); TaskCallbackImpl callback2 = new TaskCallbackImpl(listener, taskName, envelope2, coordinator2, 2, 0); List<TaskCallbackImpl> callbacksToUpdate = callbackManager.updateCallback(callback2); assertTrue(callbacksToUpdate.isEmpty()); IncomingMessageEnvelope envelope1 = new IncomingMessageEnvelope(ssp, "1", null, null); ReadableCoordinator coordinator1 = new ReadableCoordinator(taskName); coordinator1.commit(TaskCoordinator.RequestScope.CURRENT_TASK); TaskCallbackImpl callback1 = new TaskCallbackImpl(listener, taskName, envelope1, coordinator1, 1, 0); callbacksToUpdate = callbackManager.updateCallback(callback1); assertTrue(callbacksToUpdate.isEmpty()); IncomingMessageEnvelope envelope0 = new IncomingMessageEnvelope(ssp, "0", null, null); ReadableCoordinator coordinator = new ReadableCoordinator(taskName); TaskCallbackImpl callback0 = new TaskCallbackImpl(listener, taskName, envelope0, coordinator, 0, 0); callbacksToUpdate = callbackManager.updateCallback(callback0); assertEquals(2, callbacksToUpdate.size()); TaskCallbackImpl taskCallback = callbacksToUpdate.get(0); assertTrue(taskCallback.matchSeqNum(0)); assertEquals(ssp, taskCallback.getSystemStreamPartition()); assertEquals("0", taskCallback.getOffset()); taskCallback = callbacksToUpdate.get(1); assertTrue(taskCallback.matchSeqNum(1)); assertEquals(ssp, taskCallback.getSystemStreamPartition()); assertEquals("1", taskCallback.getOffset()); }
@Test public void testUpdateShouldReturnAllCompletedCallbacksTillTheCommitRequestDefined() { TaskName taskName = new TaskName("Partition 0"); SystemStreamPartition ssp1 = new SystemStreamPartition("kafka", "topic", new Partition(0)); SystemStreamPartition ssp2 = new SystemStreamPartition("kafka", "topic", new Partition(0)); IncomingMessageEnvelope envelope3 = new IncomingMessageEnvelope(ssp2, "0", null, null); ReadableCoordinator coordinator3 = new ReadableCoordinator(taskName); coordinator3.commit(TaskCoordinator.RequestScope.CURRENT_TASK); TaskCallbackImpl callback3 = new TaskCallbackImpl(listener, taskName, envelope3, coordinator3, 3, 0); List<TaskCallbackImpl> callbacksToUpdate = callbackManager.updateCallback(callback3); assertTrue(callbacksToUpdate.isEmpty()); IncomingMessageEnvelope envelope2 = new IncomingMessageEnvelope(ssp1, "2", null, null); ReadableCoordinator coordinator2 = new ReadableCoordinator(taskName); coordinator2.shutdown(TaskCoordinator.RequestScope.ALL_TASKS_IN_CONTAINER); TaskCallbackImpl callback2 = new TaskCallbackImpl(listener, taskName, envelope2, coordinator2, 2, 0); callbacksToUpdate = callbackManager.updateCallback(callback2); assertTrue(callbacksToUpdate.isEmpty()); IncomingMessageEnvelope envelope1 = new IncomingMessageEnvelope(ssp1, "1", null, null); ReadableCoordinator coordinator1 = new ReadableCoordinator(taskName); coordinator1.commit(TaskCoordinator.RequestScope.CURRENT_TASK); TaskCallbackImpl callback1 = new TaskCallbackImpl(listener, taskName, envelope1, coordinator1, 1, 0); callbacksToUpdate = callbackManager.updateCallback(callback1); assertTrue(callbacksToUpdate.isEmpty()); IncomingMessageEnvelope envelope0 = new IncomingMessageEnvelope(ssp1, "0", null, null); ReadableCoordinator coordinator = new ReadableCoordinator(taskName); TaskCallbackImpl callback0 = new TaskCallbackImpl(listener, taskName, envelope0, coordinator, 0, 0); callbacksToUpdate = callbackManager.updateCallback(callback0); assertEquals(2, callbacksToUpdate.size()); TaskCallbackImpl callback = callbacksToUpdate.get(0); assertTrue(callback.matchSeqNum(0)); assertEquals(envelope0.getSystemStreamPartition(), callback.getSystemStreamPartition()); assertEquals(envelope0.getOffset(), callback.getOffset()); callback = callbacksToUpdate.get(1); assertTrue(callback.matchSeqNum(1)); assertEquals(envelope1.getSystemStreamPartition(), callback.getSystemStreamPartition()); assertEquals(envelope1.getOffset(), callback.getOffset()); }