@Test
  /**
   * same as other happy path messages, but with one processor removed in the middle
   */ public void testStreamProcessorWithRemove() {

    // set number of events we expect to read by both processes in total:
    // p1 and p2 - both read messageCount at first and p1 is shutdown, new batch of events is generated
    // and p2 will read all of them from the beginning (+ 2 x messageCounts, total 3 x)
    int totalEventsToGenerate = 3 * messageCount;
    TestStreamTask.endLatch = new CountDownLatch(totalEventsToGenerate);

    // create first processor
    CountDownLatch waitStart1 = new CountDownLatch(1);
    CountDownLatch waitStop1 = new CountDownLatch(1);
    StreamProcessor sp1 = createStreamProcessor("30", map, waitStart1, waitStop1);

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

    // start the second processor
    CountDownLatch waitStart2 = new CountDownLatch(1);
    CountDownLatch waitStop2 = new CountDownLatch(1);
    StreamProcessor sp2 = createStreamProcessor("31", map, waitStart2, waitStop2);

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

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

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

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

    // make sure they consume all the messages from the first batch
    waitUntilMessagesLeftN(totalEventsToGenerate - messageCount);
    CountDownLatch containerStopped2 = sp2.containerShutdownLatch;

    // stop the first processor
    stopProcessor(stopLatch1);

    // wait until it's really down
    waitForProcessorToStartStop(waitStop1);

    // processor2 will stop it container and start again.
    // We wait for the container's stop to make sure we can count EXACTLY how many messages it reads.

    LOG.info("containerStopped latch = " + containerStopped2);
    waitForProcessorToStartStop(containerStopped2);

    // 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 p2 consumes all the message by itself;
    waitUntilMessagesLeftN(0);

    // shutdown p2

    try {
      stopProcessor(stopLatch2);
      t2.join(1000);
    } catch (InterruptedException e) {
      Assert.fail("Failed to join finished thread:" + e.getLocalizedMessage());
    }

    // processor1 and 2 will both read 20 events (total 40), and then processor2 read 80 events by itself,
    // but the expected values are the same 0-79 - we should get each value one time.
    // Meanwhile the number of events we gonna get is 40 + 80
    verifyNumMessages(outputTopic, 2 * messageCount, totalEventsToGenerate);
  }
Similar to the previous tests, but add another processor in the middle
samza-test/src/test/java/org/apache/samza/processor/TestZkStreamProcessor.java