@Override
  public void init() {
    if (isInitialized.compareAndSet(false, true)) {
      LOG.info("Starting the coordinator stream system consumer.");
      LOG.debug("Starting the coordinator stream system consumer with config: {}", config);
      registerConsumer();
      systemConsumer.start();
      systemProducer.register(SOURCE);
      systemProducer.start();
      iterator = new SystemStreamPartitionIterator(systemConsumer, coordinatorSystemStreamPartition);
      readMessagesFromCoordinatorStream();
    } else {
      LOG.info("Store had already been initialized. Skipping.");
    }
  }
An implementation of the {@link MetadataStore} interface where the metadata of the samza job is stored in coordinator stream.

This class is thread safe.

It is recommended to use {@link NamespaceAwareCoordinatorStreamStore}. This will enable the single CoordinatorStreamStore connection
to be shared by the multiple {@link NamespaceAwareCoordinatorStreamStore} instances.
samza-core/src/main/java/org/apache/samza/coordinator/metadatastore/CoordinatorStreamStore.java