@Test
  public void shouldReElectLeaderWhenLeaderDies() throws InterruptedException {
    // Set up kafka topics.
    publishKafkaEvents(inputKafkaTopic, 0, 2 * NUM_KAFKA_EVENTS, PROCESSOR_IDS[0]);

    // Create stream applications.
    CountDownLatch kafkaEventsConsumedLatch = new CountDownLatch(2 * NUM_KAFKA_EVENTS);
    CountDownLatch processedMessagesLatch1 = new CountDownLatch(1);
    CountDownLatch processedMessagesLatch2 = new CountDownLatch(1);
    CountDownLatch processedMessagesLatch3 = new CountDownLatch(1);

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

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

    // Wait until all processors have processed a message.
    processedMessagesLatch1.await();
    processedMessagesLatch2.await();

    // Verifications before killing the leader.
    String jobModelVersion = zkUtils.getJobModelVersion();
    JobModel jobModel = JobModelUtil.readJobModel(jobModelVersion, zkMetadataStore);
    assertEquals(2, jobModel.getContainers().size());
    assertEquals(Sets.newHashSet("0000000000", "0000000001"), jobModel.getContainers().keySet());
    assertEquals("1", jobModelVersion);

    List<String> processorIdsFromZK = zkUtils.getActiveProcessorsIDs(Arrays.asList(PROCESSOR_IDS));

    assertEquals(2, processorIdsFromZK.size());
    assertEquals(PROCESSOR_IDS[0], processorIdsFromZK.get(0));

    // Kill the leader. Since appRunner1 is the first to join the cluster, it's the leader.
    appRunner1.kill();
    appRunner1.waitForFinish();
    assertEquals(ApplicationStatus.SuccessfulFinish, appRunner1.status());

    kafkaEventsConsumedLatch.await();
    publishKafkaEvents(inputKafkaTopic, 0, 2 * NUM_KAFKA_EVENTS, PROCESSOR_IDS[0]);

    executeRun(appRunner3, applicationConfig3);
    processedMessagesLatch3.await();

    // Verifications after killing the leader.
    processorIdsFromZK = zkUtils.getActiveProcessorsIDs(ImmutableList.of(PROCESSOR_IDS[1], PROCESSOR_IDS[2]));
    assertEquals(2, processorIdsFromZK.size());
    assertEquals(PROCESSOR_IDS[1], processorIdsFromZK.get(0));
    jobModelVersion = zkUtils.getJobModelVersion();
    jobModel = JobModelUtil.readJobModel(jobModelVersion, zkMetadataStore);
    assertEquals(Sets.newHashSet("0000000001", "0000000002"), jobModel.getContainers().keySet());
    assertEquals(2, jobModel.getContainers().size());

    appRunner2.kill();
    appRunner2.waitForFinish();
    assertEquals(ApplicationStatus.SuccessfulFinish, appRunner2.status());
    appRunner3.kill();
    appRunner3.waitForFinish();
    assertEquals(ApplicationStatus.SuccessfulFinish, appRunner3.status());
  }
sspGrouper is set to AllSspToSingleTaskGrouperFactory (All ssps from input kafka topic are mapped to a single task per container).
AllSspToSingleTaskGrouperFactory should be used only with high-level consumers which do the partition management
by themselves. Using the factory with the consumers that do not do the partition management will result in
each processor/task consuming all the messages from all the partitions.
Run a stream application(streamApp1) consuming messages from input topic(effectively one container).

In the callback triggered by streamApp1 after processing a message, bring up an another stream application(streamApp2).

Assertions:
          A) JobModel generated before and after the addition of streamApp2 should not be equal.
          B) Second stream application(streamApp2) should join the group and process all the messages.
samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java