private void drain() {
      state.complete = true;
      state.startDrain();
      try {
        ReadableCoordinator coordinator = new ReadableCoordinator(task.taskName());

        task.drain(coordinator);

        // issue a shutdown request for the task
        coordinator.shutdown(TaskCoordinator.RequestScope.CURRENT_TASK);
        coordinatorRequests.update(coordinator);

        // issue a commit explicitly before we shutdown the task
        // Adding commit to coordinator will not work as the state is marked complete and NO_OP will always be the
        // next operation for this task
        commit();
      } finally {
        resumeAfterDrain();
      }
    }
Called when a task has drained i.e all SSPs for the task have received a drain message.
samza-core/src/main/java/org/apache/samza/container/RunLoop.java