@Test
  public void testFirstProcessorDiesButSameRunIdForBatch() 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));

    CountDownLatch processedMessagesLatch1 = new CountDownLatch(1);
    ApplicationRunner appRunner1 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch1, null, null,
        applicationConfig1), applicationConfig1);
    executeRun(appRunner1, applicationConfig1);

    // firt processor is up and running
    processedMessagesLatch1.await();

    LocalApplicationRunner localApplicationRunner1 = (LocalApplicationRunner) appRunner1;

    // bring up second processor
    CountDownLatch processedMessagesLatch2 = new CountDownLatch(1);
    ApplicationRunner appRunner2 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch2, null, null,
        applicationConfig2), applicationConfig2);
    executeRun(appRunner2, applicationConfig2);

    // second processor is up and running
    processedMessagesLatch2.await();
    LocalApplicationRunner localApplicationRunner2 = (LocalApplicationRunner) appRunner2;

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

    // shut down first processor
    appRunner1.kill();
    appRunner1.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 runid matches the old ones
    LocalApplicationRunner localApplicationRunner3 = (LocalApplicationRunner) appRunner3;
    assertEquals("RunId of the new processor is not the same as that of earlier processors", localApplicationRunner2.getRunId(), localApplicationRunner3.getRunId());


    appRunner2.kill();
    appRunner2.waitForFinish();
    appRunner3.kill();
    appRunner3.waitForFinish();
  }
Test if first processor dying changes the runid for new processors joining
1. bring up two processors
2. wait till they start consuimg messages
3. kill and shutdown first processor
4. bring up a new processor
5. wait till new processor starts consuming messages
6. check if new processor gets same run.id
samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java