public static void cleanup(MetadataStore metadataStore, Config config) {
    Preconditions.checkArgument(metadataStore != null, "MetadataStore cannot be null.");
    Preconditions.checkNotNull(config, "Config parameter cannot be null.");

    final ApplicationConfig applicationConfig = new ApplicationConfig(config);
    final String runId = applicationConfig.getRunId();
    final ObjectMapper objectMapper = DrainNotificationObjectMapper.getObjectMapper();
    final NamespaceAwareCoordinatorStreamStore drainMetadataStore =
        new NamespaceAwareCoordinatorStreamStore(metadataStore, DRAIN_METADATA_STORE_NAMESPACE);

    if (DrainMonitor.shouldDrain(drainMetadataStore, runId)) {
      LOG.info("Attempting to clean up DrainNotifications from the metadata-store for the current deployment {}", runId);
      drainMetadataStore.all()
          .values()
          .stream()
          .map(bytes -> {
            try {
              return objectMapper.readValue(bytes, DrainNotification.class);
            } catch (IOException e) {
              LOG.error("Unable to deserialize DrainNotification from the metadata store", e);
              throw new SamzaException(e);
            }
          })
          .filter(notification -> runId.equals(notification.getRunId()))
          .forEach(notification -> drainMetadataStore.delete(notification.getUuid().toString()));

      drainMetadataStore.flush();
      LOG.info("Successfully cleaned up DrainNotifications from the metadata-store for the current deployment {}", runId);
    } else {
      LOG.info("No DrainNotification found in the metadata-store for the current deployment {}. No need to cleanup.",
          runId);
    }
  }
Cleans up DrainNotifications for the current deployment from the underlying metadata store.
The current runId is extracted from the config.

@param metadataStore underlying metadata store
@param config Config for the job. Used to extract the runId of the job.
samza-core/src/main/java/org/apache/samza/drain/DrainUtils.java