private MetadataStore getMetadataStore() {
    String metadataStoreFactoryClass = appDesc.getConfig().get(METADATA_STORE_FACTORY_CONFIG);
    if (metadataStoreFactoryClass == null) {
      metadataStoreFactoryClass = DEFAULT_METADATA_STORE_FACTORY;
    }
    MetadataStoreFactory metadataStoreFactory =
        ReflectionUtil.getObj(metadataStoreFactoryClass, MetadataStoreFactory.class);
    return metadataStoreFactory.getMetadataStore(STREAM_CREATION_METADATA_STORE, appDesc.getConfig(), new MetricsRegistryMap());
  }
Create intermediate streams using {@link StreamManager}.
If {@link CoordinationUtils} is provided, this function will first invoke leader election, and the leader
will create the streams. All the runner processes will wait on the latch that is released after the leader finishes
stream creation.
@param planId a unique identifier representing the plan used for coordination purpose
@param intStreams list of intermediate {@link StreamSpec}s
@param streamManager the {@link StreamManager} used to create streams
samza-core/src/main/java/org/apache/samza/execution/LocalJobPlanner.java