[FLINK-20781] Avoid NPE after SourceOperator is closed.

flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java

@@ -126,6 +126,9 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
      */
     private TimestampsAndWatermarks<OUT> eventTimeLogic;
 
+    /** Indicating whether the source operator has been closed. */
+    private boolean closed;
+
     public SourceOperator(
             FunctionWithException<SourceReaderContext, SourceReader<OUT, SplitT>, Exception>
                     readerFactory,
@@ -247,9 +250,8 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
         }
         if (sourceReader != null) {
             sourceReader.close();
-            // Set the field to null so the reader won't be closed again in dispose().
-            sourceReader = null;
         }
+        closed = true;
         super.close();
     }
 
@@ -257,7 +259,7 @@ public class SourceOperator<OUT, SplitT extends SourceSplit> extends AbstractStr
     public void dispose() throws Exception {
         // We also need to close the source reader to make sure the resources
         // are released if the task does not finish normally.
-        if (sourceReader != null) {
+        if (!closed && sourceReader != null) {
             sourceReader.close();
         }
     }