public static void cleanupAll(MetadataStore metadataStore) {
    Preconditions.checkArgument(metadataStore != null, "MetadataStore cannot be null.");
    final NamespaceAwareCoordinatorStreamStore drainMetadataStore =
        new NamespaceAwareCoordinatorStreamStore(metadataStore, DRAIN_METADATA_STORE_NAMESPACE);
    LOG.info("Attempting to cleanup all DrainNotifications from the metadata-store.");
    drainMetadataStore.all()
        .keySet()
        .forEach(drainMetadataStore::delete);
    drainMetadataStore.flush();
    LOG.info("Successfully cleaned up all DrainNotifications from the metadata-store.");
  }
Cleans up all DrainNotifications irrespective of the runId.
samza-core/src/main/java/org/apache/samza/drain/DrainUtils.java