@Test
  /**
   * Similar to the previous tests, but add another processor in the middle
   */ public void testStreamProcessorWithAdd() {

    // set number of events we expect to read by both processes in total:
    // p1 - reads 'messageCount' at first
    // p1 and p2 read all messageCount together, since they start from the beginning.
    // so we expect total 3 x messageCounts
    int totalEventsToGenerate = 3 * messageCount;
    TestStreamTask.endLatch = new CountDownLatch(totalEventsToGenerate);

    // create first processor
    CountDownLatch startWait1 = new CountDownLatch(1);
    CountDownLatch stopWait1 = new CountDownLatch(1);
    StreamProcessor sp1 = createStreamProcessor("20", map, startWait1, stopWait1);

    // produce first batch of messages starting with 0
    produceMessages(0, inputTopic, messageCount);

    // start the first processor
    CountDownLatch stopLatch1 = new CountDownLatch(1);
    Thread t1 = runInThread(sp1, stopLatch1);
    t1.start();

    // wait until the processor reports that it has started
    waitForProcessorToStartStop(startWait1);

    // make sure it consumes all the messages from the first batch
    waitUntilMessagesLeftN(totalEventsToGenerate - messageCount);
    CountDownLatch containerStopped1 = sp1.containerShutdownLatch;

    // start the second processor
    CountDownLatch startWait2 = new CountDownLatch(1);
    StreamProcessor sp2 = createStreamProcessor("21", map, startWait2, null);

    CountDownLatch stopLatch2 = new CountDownLatch(1);
    Thread t2 = runInThread(sp2, stopLatch2);
    t2.start();

    // wait until 2nd processor reports that it has started
    waitForProcessorToStartStop(startWait2);

    // wait until the 1st processor reports that it has stopped its container
    LOG.info("containerStopped latch = " + containerStopped1);
    waitForProcessorToStartStop(containerStopped1);

    // read again the first batch
    waitUntilMessagesLeftN(totalEventsToGenerate - 2 * messageCount);

    // produce the second batch of the messages, starting with 'messageCount'
    produceMessages(messageCount, inputTopic, messageCount);

    // wait until all the events are consumed
    waitUntilMessagesLeftN(0);

    // shutdown both
    try {
      stopProcessor(stopLatch1);
      stopProcessor(stopLatch2);
      t1.join(1000);
      t2.join(1000);
    } catch (InterruptedException e) {
      Assert.fail("Failed to join finished threads:" + e.getLocalizedMessage());
    }

    // p1 will read messageCount events, and then p1 and p2 will read 2xmessageCount events together,
    // but the expected values are the same 0-79, they will appear in the output more then once, but we should mark then only one time.
    // total number of events we gonna get is 80+40=120
    verifyNumMessages(outputTopic, 2 * messageCount, totalEventsToGenerate);
  }
Happy path tests.
Start 1, 2, 5 processors and make sure they all consume all the events.
samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessor.java