...

3) Update the algorithm used by JM to determine whether operators in a PipelinedRegion enables checkpoint.

• 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.TODO

• 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. 

Benchmark Results

1) DataStream#CoGroup

...