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

    Map<String, String> configMap =
        buildStreamApplicationConfigMap(testStreamAppName, testStreamAppId, false, Optional.empty());
    configMap.put(TaskConfig.TASK_SHUTDOWN_MS, "0");

    configMap.put(JobConfig.PROCESSOR_ID, PROCESSOR_IDS[0]);
    Config applicationConfig1 = new MapConfig(configMap);

    configMap.put(JobConfig.PROCESSOR_ID, PROCESSOR_IDS[1]);
    Config applicationConfig2 = new MapConfig(configMap);

    // Create StreamApplication from configuration.
    CountDownLatch kafkaEventsConsumedLatch = new CountDownLatch(NUM_KAFKA_EVENTS);
    CountDownLatch processedMessagesLatch1 = new CountDownLatch(1);
    CountDownLatch processedMessagesLatch2 = 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);

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

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

    // At this stage, both the processors are running and have drained the kakfa source.
    // Trigger re-balancing phase, by manually adding a new processor.

    configMap.put(JobConfig.PROCESSOR_ID, PROCESSOR_IDS[2]);

    // Reset the task shutdown ms for 3rd application to give it ample time to shutdown cleanly
    configMap.put(TaskConfig.TASK_SHUTDOWN_MS, TASK_SHUTDOWN_MS);
    Config applicationConfig3 = new MapConfig(configMap);

    CountDownLatch processedMessagesLatch3 = new CountDownLatch(1);

    ApplicationRunner appRunner3 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
        TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch3, null, kafkaEventsConsumedLatch,
        applicationConfig3), applicationConfig3);
    executeRun(appRunner3, applicationConfig3);

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

    processedMessagesLatch3.await();
    appRunner1.waitForFinish();
    appRunner2.waitForFinish();

    /**
     * If the processing has started in the third stream processor, then other two stream processors should be stopped.
     */
    assertEquals(ApplicationStatus.UnsuccessfulFinish, appRunner1.status());
    assertEquals(ApplicationStatus.UnsuccessfulFinish, 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