boolean areAllStreamsDrained() {
    return drainStates.values().stream().allMatch(DrainState::isDrained);
  }
Checks if all streams (input SSPs) for the task has drained.
samza-core/src/main/java/org/apache/samza/operators/impl/DrainStates.java