@Test
  public void testAllProcesssorDieNewProcessorGetsNewRunIdForBatch() throws InterruptedException {
    publishKafkaEvents(inputKafkaTopic, 0, NUM_KAFKA_EVENTS, PROCESSOR_IDS[0]);

    Map<String, String> configMap =
        buildStreamApplicationConfigMap(testStreamAppName, testStreamAppId, true, Optional.empty());
    configMap.put(JobConfig.PROCESSOR_ID, PROCESSOR_IDS[0]);
    applicationConfig1 = new ApplicationConfig(new MapConfig(configMap));
    configMap.put(JobConfig.PROCESSOR_ID, PROCESSOR_IDS[1]);
    applicationConfig2 = new ApplicationConfig(new MapConfig(configMap));
    configMap.put(JobConfig.PROCESSOR_ID, PROCESSOR_IDS[2]);
    applicationConfig3 = new ApplicationConfig(new MapConfig(configMap));

    // Create StreamApplication from configuration.
    CountDownLatch processedMessagesLatch1 = new CountDownLatch(1);
    CountDownLatch processedMessagesLatch2 = new CountDownLatch(1);

    ApplicationRunner appRunner1 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch1, null, null,
        applicationConfig1), applicationConfig1);
    ApplicationRunner appRunner2 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch2, null, null,
        applicationConfig2), applicationConfig2);

    executeRun(appRunner1, applicationConfig1);
    executeRun(appRunner2, applicationConfig2);

    processedMessagesLatch1.await();
    processedMessagesLatch2.await();

    // At this stage, both the processors are running.
    // check if their runId matches

    LocalApplicationRunner localApplicationRunner1 = (LocalApplicationRunner) appRunner1;
    LocalApplicationRunner localApplicationRunner2 = (LocalApplicationRunner) appRunner2;

    assertEquals("RunId of the two processors does not match", localApplicationRunner2.getRunId(), localApplicationRunner1.getRunId());

    String oldRunId = localApplicationRunner1.getRunId().get();

    // shut down both the processors
    appRunner1.kill();
    appRunner1.waitForFinish();
    appRunner2.kill();
    appRunner2.waitForFinish();

    //Bring up a new processsor
    CountDownLatch processedMessagesLatch3 = new CountDownLatch(1);
    ApplicationRunner appRunner3 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch3, null, null,
        applicationConfig3), applicationConfig3);
    executeRun(appRunner3, applicationConfig3);
    processedMessagesLatch3.await();

    // At this stage, the new processor is running.
    // check if new processor's runId matches that of the older processors
    LocalApplicationRunner localApplicationRunner3 = (LocalApplicationRunner) appRunner3;

    assertNotEquals("RunId of the new processor same as that of old stopped processors", oldRunId, localApplicationRunner3.getRunId());

    appRunner3.kill();
    appRunner3.waitForFinish();
  }
Test one group of processors dying and a new processor coming up generates new run.id
1. bring up two processors
2. wait till they start consuimg messages
3. kill and shutdown neatly both the processors
4. bring up a new processor
5. wait till new processor starts consuming messages
6. check if new processor has new runid different from shutdown processors
samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java