static boolean shouldDrain(NamespaceAwareCoordinatorStreamStore drainMetadataStore, String runId) {
    final Optional<List<DrainNotification>> drainNotifications = readDrainNotificationMessages(drainMetadataStore);
    if (drainNotifications.isPresent()) {
      final ImmutableList<DrainNotification> filteredDrainNotifications = drainNotifications.get()
          .stream()
          .filter(notification -> runId.equals(notification.getRunId()))
          .collect(ImmutableList.toImmutableList());
      return !filteredDrainNotifications.isEmpty();
    }
    return false;
  }
One time check check to see if there are any DrainNotification messages available in the
metadata store for the current deployment.
samza-core/src/main/java/org/apache/samza/drain/DrainMonitor.java