Utils.CollectHelper is possible to throw NPE if the sink is up but not initialized when the job fails

When the sink using Utils.CollectHelper is up but not initialized and the job fails due to other operators, it is possible that the close() method of Utils.CollectHelper is called when open() is not called.

Currently Utils.CollectHelper does not deal with this case and will throw NPE.