private void testStreamProcessor(String[] processorIds) {

    // create a latch of the size equals to the number of messages
    TestZkStreamProcessorBase.TestStreamTask.endLatch = new CountDownLatch(messageCount);

    // initialize the processors
    StreamProcessor[] streamProcessors = new StreamProcessor[processorIds.length];
    // we need to know when the processor has started
    CountDownLatch[] startWait = new CountDownLatch[processorIds.length];
    for (int i = 0; i < processorIds.length; i++) {
      startWait[i] = new CountDownLatch(1);
      streamProcessors[i] = createStreamProcessor(processorIds[i], map, startWait[i], null);
    }


    // run the processors in separate threads
    Thread[] threads = new Thread[processorIds.length];
    CountDownLatch[] stopLatches = new CountDownLatch[processorIds.length];
    for (int i = 0; i < processorIds.length; i++) {
      stopLatches[i] = new CountDownLatch(1);
      threads[i] = runInThread(streamProcessors[i], stopLatches[i]);
      threads[i].start();
    }
    // wait until the processor reports that it has started
    for (int i = 0; i < processorIds.length; i++) {
      waitForProcessorToStartStop(startWait[i]);
    }

    // produce messageCount messages, starting with key 0
    produceMessages(0, inputTopic, messageCount);

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

    // collect all the threads
    try {
      for (int i = 0; i < threads.length; i++) {
        stopProcessor(stopLatches[i]);
        threads[i].join(1000);
      }
    } catch (InterruptedException e) {
      Assert.fail("Failed to join finished thread:" + e.getLocalizedMessage());
    }

    verifyNumMessages(outputTopic, messageCount, messageCount);
  }
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