rkhachatryan commented on a change in pull request #10446: [FLINK-15076][task]
Fix SourceStreamTask cancellation
URL: https://github.com/apache/flink/pull/10446#discussion_r355482653
##########
File path:
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
##########
@@ -239,6 +244,193 @@ 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);
+ }
+
+ /**
+ * Note that this test is testing also for the shared cancellation
logic inside {@link StreamTask}
+ * which, as of the time this test is being written, is not tested
anywhere else
+ * (like {@link StreamTaskTest} or {@link OneInputStreamTaskTest}).
+ */
+ 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.awaitRunning();
+ if (withPendingMail) {
+ // This pending mail should be blocked on
checkpointLock acquisition, blocking the
+ // mailbox (task) thread.
+
testHarness.getTask().getMailboxExecutorFactory().createExecutor(0).execute(
+ () -> assertFalse(
+ "This should never execute before task
cancelation",
+ testHarness.getTask().isRunning()),
+ "Test");
+ }
+
+ try {
+ testHarness.getTask().cancel();
+ }
+ catch (ExpectedTestException e) {
+ if (!throwOnCancel) {
+ throw e;
Review comment:
`if !throwOnCancel then throw` sounds strange.
`throwOnCancel` -> `throwInCancel`?
----------------------------------------------------------------
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:
us...@infra.apache.org
With regards,
Apache Git Services
