private MapConfig getConfigFromCoordinatorStream(Config config) {
    MetadataStoreFactory metadataStoreFactory =
        ReflectionUtil.getObj(new JobConfig(config).getMetadataStoreFactory(), MetadataStoreFactory.class);
    MetadataStore metadataStore = metadataStoreFactory.getMetadataStore("set-config", config, new MetricsRegistryMap());
    metadataStore.init();
    Map<String, String> configMap = new HashMap<>();
    CoordinatorStreamValueSerde jsonSerde = new CoordinatorStreamValueSerde("set-config");
    metadataStore.all().forEach((key, value) -> {
      CoordinatorStreamStore.CoordinatorMessageKey coordinatorMessageKey = CoordinatorStreamStore.deserializeCoordinatorMessageKeyFromJson(key);
      String deserializedValue = jsonSerde.fromBytes(value);
      configMap.put(coordinatorMessageKey.getKey(), deserializedValue);
    });
    return new MapConfig(configMap);
  }
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