private void testStreamProcessorWithSessionRestart(String[] processorIds) {

    // set shorter session expiration for the test
    map.put(ZkConfig.ZK_SESSION_TIMEOUT_MS, "500");
    map.put(JobConfig.JOB_DEBOUNCE_TIME_MS, "3000");

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

    // initialize the processors
    StreamProcessor[] streamProcessors = new StreamProcessor[processorIds.length];
    ZkJobCoordinator[] jobCoordinators = new ZkJobCoordinator[processorIds.length];
    Thread[] threads = new Thread[processorIds.length];
    CountDownLatch[] threadStopLatches = new CountDownLatch[processorIds.length];
    CountDownLatch[] containerStopLatches = new CountDownLatch[processorIds.length];
    // we need to know when the processor has started
    CountDownLatch[] startWait = new CountDownLatch[processorIds.length];
    //CountDownLatch[] stopWait = 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);
      jobCoordinators[i] = (ZkJobCoordinator) streamProcessors[i].getCurrentJobCoordinator();
    }

    // run the processors in separate threads
    for (int i = 0; i < processorIds.length; i++) {
      threadStopLatches[i] = new CountDownLatch(1); // is used int stopProcessor
      threads[i] = runInThread(streamProcessors[i], threadStopLatches[i]);
      threads[i].start();
    }

    for (int i = 0; i < processorIds.length; i++) {
      // wait until the processor reports that it has started
      waitForProcessorToStartStop(startWait[i]);
    }

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

    // make sure it consumes all the messages from the first batch
    waitUntilMessagesLeftN(totalEventsToGenerate - messageCount);

    // Get the container stop latch to be able to check when a container is stopped.
    // New containerShutdownLatch is created after each onNewJobModel,
    // so we need to get the current one, before it changed..
    for (int i = 0; i < processorIds.length; i++) {
      containerStopLatches[i] = streamProcessors[i].containerShutdownLatch;
    }

    // expire zk session of one of the processors
    expireSession(jobCoordinators[0].getZkUtils().getZkClient());

    // wait until all other processors report that they have stopped their containers
    for (int i = 0; i < processorIds.length; i++) {
      waitForProcessorToStartStop(containerStopLatches[i]);
    }

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

    produceMessages(messageCount, inputTopic, messageCount);

    waitUntilMessagesLeftN(0);

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

    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/TestZkStreamProcessorSession.java