...

• If the PipelinedRegion contain any operator with getHasInternalSorter() == true, then the checkpoint is disabled for this region.
• If the PipelineRegion contains no sink operator (i.e. operators without out-edges), then the checkpoint is disabled for this region.

4) Allow a pipelined region to read from the blocking edge while still achieving exactly-once semantics with checkpoint enabled.CheckpointCoordinator will trigger checkpoint only after all ExecutionVertexes have finished reading the blocking input edges. 

• The JM should keep track of the number of ExecutionVertexes which have blocking input edges that have not been fully read.
• Each ExecutionVertex should notify JM after it finishes reading all its blocking input edges.
• The JM should enable checkpoint only after all ExecutionVertexes have finished reading the blocking input edges.
• Records buffered in the upstream node due to blocking edges can be removed after its downstream nodes have made the first checkpoint. 

5) An input edge with pending records is same as a source with isBacklog=true when an operator determines its RecordAttributes for downstream nodes.

Benchmark Results

1) DataStream#CoGroup

...