boolean isDrained(SystemStream systemStream) {
    return drainStates.entrySet().stream()
        .filter(entry -> entry.getKey().getSystemStream().equals(systemStream))
        .allMatch(entry -> entry.getValue().isDrained());
  }
Checks if the system-stream is drained.
samza-core/src/main/java/org/apache/samza/operators/impl/DrainStates.java