[FLINK-19040][task] Close SourceReader in SourceOperator

zhengcanbin · Aug 25, 2020 · f666b81 · f666b81
1 parent d4970e2
commit f666b81
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 0 deletions.
diff --git a/...streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java b/...streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
@@ -162,6 +162,7 @@ public void sendSourceEventToCoordinator(SourceEvent event) {
 
 	@Override
 	public void close() throws Exception {
+		sourceReader.close();
 		eventTimeLogic.stopPeriodicWatermarkEmits();
 		super.close();
 	}

diff --git a/...aming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java b/...aming-java/src/test/java/org/apache/flink/streaming/api/operators/SourceOperatorTest.java
@@ -101,6 +101,7 @@ public void testOpen() throws Exception {
 		finally {
 			operator.close();
 		}
+		assertTrue(mockSourceReader.isClosed());
 	}
 
 	@Test
-Original file line number
+Diff line change
@@ Expand Up / @@ -101,6 +101,7 @@ public void testOpen() throws Exception { @@
     		finally {
     			operator.close();
     		}
+    		assertTrue(mockSourceReader.isClosed());
     	}
     	@Test
@@ Expand Down @@