private void readMessagesFromCoordinatorStream() {
    synchronized (bootstrapLock) {
      while (iterator.hasNext()) {
        IncomingMessageEnvelope envelope = iterator.next();
        byte[] keyAsBytes = (byte[]) envelope.getKey();
        Serde<List<?>> serde = new JsonSerde<>();
        Object[] keyArray = serde.fromBytes(keyAsBytes).toArray();
        CoordinatorStreamMessage coordinatorStreamMessage = new CoordinatorStreamMessage(keyArray, new HashMap<>());
        String namespacedKey = serializeCoordinatorMessageKeyToJson(coordinatorStreamMessage.getType(), coordinatorStreamMessage.getKey());
        if (envelope.getMessage() != null) {
          messagesReadFromCoordinatorStream.put(namespacedKey, (byte[]) envelope.getMessage());
        } else {
          messagesReadFromCoordinatorStream.remove(namespacedKey);
        }
      }
    }
  }
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