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

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

    Map<String, String> configMap =
        buildStreamApplicationConfigMap(testStreamAppName, testStreamAppId, false, Optional.of("test-store-partition-count-change"));
    configMap.put(JobConfig.PROCESSOR_ID, PROCESSOR_IDS[0]);
    ApplicationConfig applicationConfig = new ApplicationConfig(new MapConfig(configMap));

    ApplicationRunner appRunner1 = ApplicationRunners.getApplicationRunner(TestStreamApplication.getInstance(
            TEST_SYSTEM, ImmutableList.of(inputKafkaTopic), outputKafkaTopic, processedMessagesLatch1, null, kafkaEventsConsumedLatch1,
            applicationConfig), applicationConfig);

    executeRun(appRunner1, applicationConfig);
    processedMessagesLatch1.await();

    String jobModelVersion = zkUtils.getJobModelVersion();
    JobModel jobModel = JobModelUtil.readJobModel(jobModelVersion, zkMetadataStore);
    List<SystemStreamPartition> ssps = getSystemStreamPartitions(jobModel);

    // Validate that the input partition count is 5 in the JobModel.
    Assert.assertEquals(5, ssps.size());

    // Increase the partition count of input kafka topic to 100.
    increasePartitionsTo(inputKafkaTopic, 100);

    long jobModelWaitTimeInMillis = 10;
    while (Objects.equals(zkUtils.getJobModelVersion(), jobModelVersion)) {
      LOGGER.info("Waiting for new jobModel to be published");
      Thread.sleep(jobModelWaitTimeInMillis);
      jobModelWaitTimeInMillis = jobModelWaitTimeInMillis * 2;
    }

    String newJobModelVersion = zkUtils.getJobModelVersion();
    JobModel newJobModel = JobModelUtil.readJobModel(newJobModelVersion, zkMetadataStore);
    ssps = getSystemStreamPartitions(newJobModel);

    // Validate that the input partition count is 100 in the new JobModel.
    Assert.assertEquals(100, ssps.size());

    // Validate that configuration is stored in coordinator stream.
    MapConfig config = getConfigFromCoordinatorStream(applicationConfig);

    // Execution plan and serialized DAG of a samza job is stored in the config of coordinator stream. Thus, direct equals comparison between
    // the application configuration and the coordinator config will fail. Iterating through the entire configuration bag and verify that expected
    // configuration is present in the coordinator configuration.
    for (Map.Entry<String, String> entry : applicationConfig.entrySet()) {
      Assert.assertTrue(config.containsKey(entry.getKey()));
    }
  }
A. Create a kafka topic with partition count set to 5.
B. Create and launch a samza application which consumes events from the kafka topic.
C. Validate that the {@link JobModel} contains 5 {@link SystemStreamPartition}'s.
D. Increase the partition count of the input kafka topic to 100.
E. Validate that the new {@link JobModel} contains 100 {@link SystemStreamPartition}'s.
samza-test/src/test/java/org/apache/samza/test/processor/TestZkLocalApplicationRunner.java