@Test
  public void testExceptionIfInputOperatorMissing() throws NoSuchFieldException, IllegalAccessException {
    IncomingMessageEnvelope mockIme = mock(IncomingMessageEnvelope.class, RETURNS_DEEP_STUBS);
    SystemStream testSystemStream = new SystemStream("foo", "bar");
    when(mockIme.getSystemStreamPartition().getSystemStream()).thenReturn(testSystemStream);

    OperatorImplGraph mockOperatorImplGraph = mock(OperatorImplGraph.class);
    when(mockOperatorImplGraph.getInputOperator(anyObject())).thenReturn(null);
    StreamOperatorTask operatorTask = new StreamOperatorTask(mock(OperatorSpecGraph.class));
    operatorTask.setOperatorImplGraph(mockOperatorImplGraph);
    TaskCallback mockTaskCallback = mock(TaskCallback.class);
    operatorTask.processAsync(mockIme, mock(MessageCollector.class), mock(TaskCoordinator.class), mockTaskCallback);

    ArgumentCaptor<Throwable> throwableCaptor = ArgumentCaptor.forClass(Throwable.class);
    verify(mockTaskCallback, only()).failure(throwableCaptor.capture());
    assertEquals(throwableCaptor.getValue().getClass(), SamzaException.class);
    String expectedErrMessage = String.format("InputOperator not found in OperatorGraph for %s. The available input"
        + " operators are: %s. Please check SystemStream configuration for the `SystemConsumer` and/or task.inputs"
        + " task configuration.", testSystemStream, mockOperatorImplGraph.getAllInputOperators());
    assertEquals(throwableCaptor.getValue().getMessage(), expectedErrMessage);
  }
Tests if the appropriate SamzaException is propagated to the TaskCallback if there is no InputOperator for a given
SystemStream in the OperatorGraph.
samza-core/src/test/java/org/apache/samza/task/TestStreamOperatorTask.java