The SourceOperatorStreamTask does not implement cancelTask and finishTask at this point. This causes resource leak on job cancellation of finish.
Currently there are three cases that a Job may exit.
- Canceled - cancelTask() method will be called to close the SourceOperator.
- Stopped with a savepoint - finishTask() method will be called to close the SourceOperator.
- Runs to the EndOfInput - the SourceOperator is closed in StreamTask.afterInvoke().