...

2) Update Transformation subclasses (e.g. OneInputTransformation and TwoInputTransformation) to override the newly added methods using the OperatorAttributes obtained from the underlying Operator.

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

Benchmark Results

1) DataStream#CoGroup

...