Generalize the checkpoint protocol of OperatorCoordinator.

*Problem statement*

Currently, the JM uses OperatorEventValve to control the communication from OperatorCoordinator (OC) to the subtasks in order to achieve state consistency across OC and subtasks after recovering from a checkpoint. The valve is closed when a checkpoint starts and reopened after the checkpoint barriers are sent to the source subtasks.

While this mechanism works for the source operators, it unnecessarily limits the general usage of OC due to the following limitations:

• It does not handle (e.g. blocks) the control flow messages from subtasks to OC.
• It does not handle the control flow messages from OC to subtasks which are sent after checkpoint barriers have been sent to sources but before subtasks have received those barriers.

If the limitations mentioned above are not satisfied, consistency issues might occur. For example, if a subtask sends messages to its coordinator after the checkpoint barriers are sent to the sources and before the subtasks receive the barriers, these messages would be recorded in the subtask's snapshot but not the coordinator's. When the Flink job recovers from this snapshot, the subtask would have a record of sending out the message while the coordinator has no record of receiving it.

*Proposed solution*

We plan to address this problem by extending the blocking period. The communication should be blocked before the OC starts the checkpoint and reopened for each individual subtask after that subtask finishes the checkpoint, to make sure that both OCs and subtasks would see the same version of message after recovering from a checkpoint. Communications in both directions should be blocked during this period. And the messages blocked on the subtasks side should be properly stored in the checkpoint snapshot for recovery.