[GitHub] [flink] rkhachatryan commented on a change in pull request #10446: [FLINK-15076][task] Fix SourceStreamTask cancellation

[GitBox <gi...@apache.org>]

rkhachatryan commented on a change in pull request #10446: [FLINK-15076][task] Fix SourceStreamTask cancellation
URL: https://github.com/apache/flink/pull/10446#discussion_r354422240
 
 

 ##########
 File path: flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
 ##########
 @@ -239,6 +243,128 @@ public void testNotMarkingEndOfInputWhenTaskCancelled () throws Exception {
 			testHarness.getOutput());
 	}
 
+	@Test
+	public void testCancellationWithSourceBlockedOnLock() throws Exception {
+		testCancellationWithSourceBlockedOnLock(false, false);
+	}
+
+	@Test
+	public void testCancellationWithSourceBlockedOnLockWithPendingMail() throws Exception {
+		testCancellationWithSourceBlockedOnLock(true, false);
+	}
+
+	@Test
+	public void testCancellationWithSourceBlockedOnLockAndThrowingOnError() throws Exception {
+		testCancellationWithSourceBlockedOnLock(false, true);
+	}
+
+	@Test
+	public void testCancellationWithSourceBlockedOnLockWithPendingMailAndThrowingOnError() throws Exception {
+		testCancellationWithSourceBlockedOnLock(true, true);
+	}
+
+	public void testCancellationWithSourceBlockedOnLock(boolean withPendingMail, boolean throwOnCancel) throws Exception {
+		final StreamTaskTestHarness<String> testHarness = new StreamTaskTestHarness<>(
+			SourceStreamTask::new,
+			BasicTypeInfo.STRING_TYPE_INFO);
+
+		CancelLockingSource.reset();
+		testHarness
+			.setupOperatorChain(
+				new OperatorID(),
+				new StreamSource<>(new CancelLockingSource(throwOnCancel)))
+			.chain(
+				new OperatorID(),
+				new TestBoundedOneInputStreamOperator("Operator1"),
+				BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()))
+			.finish();
+
+		StreamConfig streamConfig = testHarness.getStreamConfig();
+		streamConfig.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);
+
+		testHarness.invoke();
+		CancelLockingSource.getDataProcessing().get();
 
 Review comment:
   I'd encapsulate this into something like `awaitCompletion()`.

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services