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

    // Create a TaskApplication with only one task per container.
    // The task does not invokes taskCallback.complete for any of the dispatched message.
    CountDownLatch shutdownLatch = new CountDownLatch(1);
    CountDownLatch processedMessagesLatch1 = new CountDownLatch(1);

    TaskApplication taskApplication = new TestTaskApplication(TEST_SYSTEM, inputKafkaTopic, outputKafkaTopic, processedMessagesLatch1, shutdownLatch);
    MapConfig taskApplicationConfig = new MapConfig(ImmutableList.of(applicationConfig1,
        ImmutableMap.of(TaskConfig.MAX_CONCURRENCY, "1", JobConfig.SSP_GROUPER_FACTORY, "org.apache.samza.container.grouper.stream.AllSspToSingleTaskGrouperFactory")));
    ApplicationRunner appRunner = ApplicationRunners.getApplicationRunner(taskApplication, taskApplicationConfig);

    // Run the application.
    executeRun(appRunner, applicationConfig1);

    // Wait for the task to receive at least one dispatched message.
    processedMessagesLatch1.await();

    // Kill the application when none of the dispatched messages is acknowledged as completed by the task.
    appRunner.kill();
    appRunner.waitForFinish();

    // Expect the shutdown latch to be triggered.
    shutdownLatch.await();

    // Assert that the shutdown was successful.
    Assert.assertEquals(ApplicationStatus.SuccessfulFinish, appRunner.status());
  }
A. Create a input kafka topic: T1 with partition count set to 32.
B. Create a input kafka topic: T2 with partition count set to 32.
C. Create and launch a stateful samza application which consumes events from the kafka topics: T1 and T2.
D. Validate that the {@link JobModel} contains 32 {@link SystemStreamPartition}'s of T1 and 32 {@link SystemStreamPartition}'s of T2.
E. Increase the partition count of the input kafka topic: T1 to 64.
F. Increase the partition count of the input kafka topic: T2 to 64.
G. Validate that the new {@link JobModel} contains 64 {@link SystemStreamPartition}'s of T1, T2 and the input
SystemStreamPartitions are mapped to the correct task.
samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java