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:
[email protected]
With regards,
Apache Git Services
