@Test
  public void testExceptionsInProcessInvokesTaskCallback() throws InterruptedException {
    ExecutorService taskThreadPool = Executors.newFixedThreadPool(2);
    TaskCallback mockTaskCallback = mock(TaskCallback.class);
    MessageCollector mockMessageCollector = mock(MessageCollector.class);
    TaskCoordinator mockTaskCoordinator = mock(TaskCoordinator.class);
    StreamOperatorTask operatorTask = new StreamOperatorTask(mock(OperatorSpecGraph.class));
    operatorTask.setTaskThreadPool(taskThreadPool);

    CountDownLatch failureLatch = new CountDownLatch(1);

    doAnswer(ctx -> {
      failureLatch.countDown();
      return null;
    }).when(mockTaskCallback).failure(anyObject());

    operatorTask.processAsync(mock(IncomingMessageEnvelope.class), mockMessageCollector,
        mockTaskCoordinator, mockTaskCallback);
    failureLatch.await();
  }
Pass an invalid IME to processAsync. Any exceptions in processAsync should still get propagated through the
task callback.
samza-core/src/test/java/org/apache/samza/task/TestStreamOperatorTask.java