@VisibleForTesting
  static UUID writeDrainNotification(MetadataStore metadataStore, DrainNotification drainNotification) {
    Preconditions.checkArgument(metadataStore != null, "MetadataStore cannot be null.");
    Preconditions.checkArgument(drainNotification != null, "DrainNotification cannot be null.");
    final NamespaceAwareCoordinatorStreamStore drainMetadataStore =
        new NamespaceAwareCoordinatorStreamStore(metadataStore, DRAIN_METADATA_STORE_NAMESPACE);
    final ObjectMapper objectMapper = DrainNotificationObjectMapper.getObjectMapper();
    try {
      drainMetadataStore.put(drainNotification.getUuid().toString(), objectMapper.writeValueAsBytes(drainNotification));
      drainMetadataStore.flush();
      LOG.info("DrainNotification with id {} written to metadata-store for the deployment ID {}", drainNotification.getUuid(), drainNotification.getUuid());
    } catch (Exception ex) {
      throw new SamzaException(
          String.format("DrainNotification might have been not written to metastore %s", drainNotification), ex);
    }
    return drainNotification.getUuid();
  }
Writes a {@link DrainNotification} to an underlying metadata store.
samza-core/src/main/java/org/apache/samza/drain/DrainUtils.java