You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2019/12/10 10:34:40 UTC
[flink] branch release-1.10 updated (52a2f03 -> 1e723c6)
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a change to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 52a2f03 [FLINK-15166][runtime] Fix the bug of wrongly recycling uncompressed buffer
new f948707 [hotfix][test] Do not hide original exception in the SourceStreamTaskTest
new 94fa197 [FLINK-15076][task] Fix SourceStreamTask cancellation
new a617aee [hotfix][test] Rename and reformat testCancellationFailsWithBlockingLock
new 1e723c6 [hotfix][task] Do not wrap Errors from SourceFunctionThread
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../streaming/runtime/tasks/SourceStreamTask.java | 10 +-
.../flink/streaming/runtime/tasks/StreamTask.java | 34 ++-
.../streaming/runtime/tasks/mailbox/Mail.java | 7 +-
.../runtime/tasks/mailbox/MailboxProcessor.java | 10 +-
.../runtime/tasks/SourceStreamTaskTest.java | 192 +++++++++++++++-
.../streaming/runtime/tasks/StreamTaskTest.java | 247 ++++++++-------------
.../tasks/mailbox/MailboxExecutorImplTest.java | 13 --
.../tasks/mailbox/TaskMailboxProcessorTest.java | 5 +-
8 files changed, 332 insertions(+), 186 deletions(-)
[flink] 03/04: [hotfix][test] Rename and reformat
testCancellationFailsWithBlockingLock
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a617aeedf9a3db5bee03b18bf0147a9613a9268c
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Thu Dec 5 16:23:17 2019 +0100
[hotfix][test] Rename and reformat testCancellationFailsWithBlockingLock
New name testCanceleablesCanceledOnCancelTaskError better reflects what the test is doing
while CancelFailingTask closer to the test case makes reading the test easier.
---
.../streaming/runtime/tasks/StreamTaskTest.java | 188 ++++++++++-----------
1 file changed, 94 insertions(+), 94 deletions(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
index 7719259..9917301 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
@@ -315,7 +315,7 @@ public class StreamTaskTest extends TestLogger {
}
@Test
- public void testCancellationFailsWithBlockingLock() throws Exception {
+ public void testCanceleablesCanceledOnCancelTaskError() throws Exception {
syncLatch = new OneShotLatch();
StreamConfig cfg = new StreamConfig(new Configuration());
@@ -334,6 +334,99 @@ public class StreamTaskTest extends TestLogger {
assertEquals(ExecutionState.CANCELED, task.getExecutionState());
}
+ /**
+ * A task that locks for ever, fail in {@link #cancelTask()}. It can be only shut down cleanly
+ * if {@link StreamTask#getCancelables()} are closed properly.
+ */
+ public static class CancelFailingTask extends StreamTask<String, AbstractStreamOperator<String>> {
+
+ public CancelFailingTask(Environment env) {
+ super(env);
+ }
+
+ @Override
+ protected void init() {}
+
+ @Override
+ protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {
+ final OneShotLatch latch = new OneShotLatch();
+ final Object lock = new Object();
+
+ LockHolder holder = new LockHolder(lock, latch);
+ holder.start();
+ try {
+ // cancellation should try and cancel this
+ getCancelables().registerCloseable(holder);
+
+ // wait till the lock holder has the lock
+ latch.await();
+
+ // we are at the point where cancelling can happen
+ syncLatch.trigger();
+
+ // try to acquire the lock - this is not possible as long as the lock holder
+ // thread lives
+ //noinspection SynchronizationOnLocalVariableOrMethodParameter
+ synchronized (lock) {
+ // nothing
+ }
+ }
+ finally {
+ holder.close();
+ }
+ controller.allActionsCompleted();
+ }
+
+ @Override
+ protected void cleanup() {}
+
+ @Override
+ protected void cancelTask() throws Exception {
+ throw new Exception("test exception");
+ }
+
+ /**
+ * A thread that holds a lock as long as it lives.
+ */
+ private static final class LockHolder extends Thread implements Closeable {
+
+ private final OneShotLatch trigger;
+ private final Object lock;
+ private volatile boolean canceled;
+
+ private LockHolder(Object lock, OneShotLatch trigger) {
+ this.lock = lock;
+ this.trigger = trigger;
+ }
+
+ @Override
+ public void run() {
+ synchronized (lock) {
+ while (!canceled) {
+ // signal that we grabbed the lock
+ trigger.trigger();
+
+ // basically freeze this thread
+ try {
+ //noinspection SleepWhileHoldingLock
+ Thread.sleep(1000000000);
+ } catch (InterruptedException ignored) {}
+ }
+ }
+ }
+
+ public void cancel() {
+ canceled = true;
+ }
+
+ @Override
+ public void close() {
+ canceled = true;
+ interrupt();
+ }
+ }
+ }
+
@Test
public void testDecliningCheckpointStreamOperator() throws Exception {
DeclineDummyEnvironment declineDummyEnvironment = new DeclineDummyEnvironment();
@@ -1360,58 +1453,6 @@ public class StreamTaskTest extends TestLogger {
}
}
- /**
- * A task that locks if cancellation attempts to cleanly shut down.
- */
- public static class CancelFailingTask extends StreamTask<String, AbstractStreamOperator<String>> {
-
- public CancelFailingTask(Environment env) {
- super(env);
- }
-
- @Override
- protected void init() {}
-
- @Override
- protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {
- final OneShotLatch latch = new OneShotLatch();
- final Object lock = new Object();
-
- LockHolder holder = new LockHolder(lock, latch);
- holder.start();
- try {
- // cancellation should try and cancel this
- getCancelables().registerCloseable(holder);
-
- // wait till the lock holder has the lock
- latch.await();
-
- // we are at the point where cancelling can happen
- syncLatch.trigger();
-
- // try to acquire the lock - this is not possible as long as the lock holder
- // thread lives
- //noinspection SynchronizationOnLocalVariableOrMethodParameter
- synchronized (lock) {
- // nothing
- }
- }
- finally {
- holder.close();
- }
- controller.allActionsCompleted();
- }
-
- @Override
- protected void cleanup() {}
-
- @Override
- protected void cancelTask() throws Exception {
- throw new Exception("test exception");
- }
-
- }
-
private static class ThreadInspectingTask extends StreamTask<String, AbstractStreamOperator<String>> {
private final long taskThreadId;
@@ -1482,47 +1523,6 @@ public class StreamTaskTest extends TestLogger {
// ------------------------------------------------------------------------
// ------------------------------------------------------------------------
- /**
- * A thread that holds a lock as long as it lives.
- */
- private static final class LockHolder extends Thread implements Closeable {
-
- private final OneShotLatch trigger;
- private final Object lock;
- private volatile boolean canceled;
-
- private LockHolder(Object lock, OneShotLatch trigger) {
- this.lock = lock;
- this.trigger = trigger;
- }
-
- @Override
- public void run() {
- synchronized (lock) {
- while (!canceled) {
- // signal that we grabbed the lock
- trigger.trigger();
-
- // basically freeze this thread
- try {
- //noinspection SleepWhileHoldingLock
- Thread.sleep(1000000000);
- } catch (InterruptedException ignored) {}
- }
- }
- }
-
- public void cancel() {
- canceled = true;
- }
-
- @Override
- public void close() {
- canceled = true;
- interrupt();
- }
- }
-
static class TestStreamSource<OUT, SRC extends SourceFunction<OUT>> extends StreamSource<OUT, SRC> {
static AbstractKeyedStateBackend<?> keyedStateBackend;
[flink] 04/04: [hotfix][task] Do not wrap Errors from
SourceFunctionThread
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1e723c64adc962a5fc054ce3ee7646fe51cfbf44
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Dec 9 18:24:42 2019 +0100
[hotfix][task] Do not wrap Errors from SourceFunctionThread
---
.../apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
index 05e968c..81d39ed 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
@@ -198,6 +198,9 @@ public class MailboxProcessor implements Closeable {
if (throwable instanceof Exception) {
throw (Exception) throwable;
}
+ else if (throwable instanceof Error) {
+ throw (Error) throwable;
+ }
else {
throw WrappingRuntimeException.wrapIfNecessary(throwable);
}
[flink] 01/04: [hotfix][test] Do not hide original exception in the
SourceStreamTaskTest
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit f9487076c8633019ef62e3f888411a0905d44b0b
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Thu Dec 5 15:31:48 2019 +0100
[hotfix][test] Do not hide original exception in the SourceStreamTaskTest
---
.../apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
index 6a46a86..87a3a47 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
@@ -229,7 +229,9 @@ public class SourceStreamTaskTest {
try {
testHarness.waitForTaskCompletion();
} catch (Throwable t) {
- assertTrue(ExceptionUtils.findThrowable(t, CancelTaskException.class).isPresent());
+ if (!ExceptionUtils.findThrowable(t, CancelTaskException.class).isPresent()) {
+ throw t;
+ }
}
expectedOutput.add(new StreamRecord<>("Hello"));
[flink] 02/04: [FLINK-15076][task] Fix SourceStreamTask cancellation
Posted by pn...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 94fa197f92537d608f2354aff803dbda340ec92f
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Dec 6 20:51:48 2019 +0100
[FLINK-15076][task] Fix SourceStreamTask cancellation
Source thread should be interrupted more or less the same way how task thread is
being interrupted. This is important for example as in the scenario presented in the
SourceStreamTaskTest#testCancellationWithSourceBlockedOnLock(). SourceFunction is
blocked while holding checkpointLock, which might prevent task thread from
cancelling properly if the SourceFunction is not interrupted.
---
.../streaming/runtime/tasks/SourceStreamTask.java | 10 +-
.../flink/streaming/runtime/tasks/StreamTask.java | 34 +++-
.../streaming/runtime/tasks/mailbox/Mail.java | 7 +-
.../runtime/tasks/mailbox/MailboxProcessor.java | 7 +-
.../runtime/tasks/SourceStreamTaskTest.java | 188 +++++++++++++++++++++
.../streaming/runtime/tasks/StreamTaskTest.java | 69 --------
.../tasks/mailbox/MailboxExecutorImplTest.java | 13 --
.../tasks/mailbox/TaskMailboxProcessorTest.java | 5 +-
8 files changed, 237 insertions(+), 96 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
index 3507bb5..2e11548 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTask.java
@@ -131,8 +131,13 @@ public class SourceStreamTask<OUT, SRC extends SourceFunction<OUT>, OP extends S
@Override
protected void cancelTask() {
- if (headOperator != null) {
- headOperator.cancel();
+ try {
+ if (headOperator != null) {
+ headOperator.cancel();
+ }
+ }
+ finally {
+ sourceThread.interrupt();
}
}
@@ -191,6 +196,7 @@ public class SourceStreamTask<OUT, SRC extends SourceFunction<OUT>, OP extends S
headOperator.run(getCheckpointLock(), getStreamStatusMaintainer(), operatorChain);
completionFuture.complete(null);
} catch (Throwable t) {
+ // Note, t can be also an InterruptedException
completionFuture.completeExceptionally(t);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
index da83dc8..aeac7da 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
@@ -76,6 +76,7 @@ import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailbox;
import org.apache.flink.streaming.runtime.tasks.mailbox.TaskMailboxImpl;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingRuntimeException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -466,7 +467,7 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
// let the task do its work
isRunning = true;
- mailboxProcessor.runMailboxLoop();
+ runMailboxLoop();
// if this left the run() method cleanly despite the fact that this was canceled,
// make sure the "clean shutdown" is not attempted
@@ -481,6 +482,37 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
}
}
+ private void runMailboxLoop() throws Exception {
+ try {
+ try {
+ mailboxProcessor.runMailboxLoop();
+ }
+ catch (WrappingRuntimeException wrappingException) {
+ Throwable unwrapped = wrappingException.unwrap();
+ if (unwrapped instanceof Exception) {
+ throw (Exception) unwrapped;
+ }
+ else {
+ throw wrappingException;
+ }
+ }
+ }
+ catch (InterruptedException e) {
+ if (!canceled) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ }
+ catch (Exception e) {
+ if (canceled) {
+ LOG.warn("Error while canceling task.", e);
+ }
+ else {
+ throw e;
+ }
+ }
+ }
+
private void afterInvoke() throws Exception {
LOG.debug("Finished task {}", getName());
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java
index 8665a4a..db1c1c4 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/Mail.java
@@ -19,7 +19,6 @@ package org.apache.flink.streaming.runtime.tasks.mailbox;
import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor;
-import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.function.RunnableWithException;
@@ -76,10 +75,6 @@ public class Mail {
}
public void run() throws Exception {
- try {
- actionExecutor.run(runnable);
- } catch (Exception e) {
- throw new FlinkException("Cannot process mail " + toString(), e);
- }
+ actionExecutor.run(runnable);
}
}
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
index 0c93ae3..05e968c 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxProcessor.java
@@ -195,7 +195,12 @@ public class MailboxProcessor implements Closeable {
public void reportThrowable(Throwable throwable) {
sendControlMail(
() -> {
- throw WrappingRuntimeException.wrapIfNecessary(throwable);
+ if (throwable instanceof Exception) {
+ throw (Exception) throwable;
+ }
+ else {
+ throw WrappingRuntimeException.wrapIfNecessary(throwable);
+ }
},
"Report throwable %s", throwable);
}
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
index 87a3a47..4e02a86 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/SourceStreamTaskTest.java
@@ -29,6 +29,7 @@ import org.apache.flink.runtime.checkpoint.CheckpointMetaData;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.execution.CancelTaskException;
import org.apache.flink.runtime.jobgraph.OperatorID;
+import org.apache.flink.runtime.operators.testutils.ExpectedTestException;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.checkpoint.ListCheckpointed;
import org.apache.flink.streaming.api.functions.source.FromElementsFunction;
@@ -54,6 +55,7 @@ import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -62,6 +64,7 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicLong;
import static org.apache.flink.util.Preconditions.checkState;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
/**
@@ -241,6 +244,191 @@ public class SourceStreamTaskTest {
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 throwInCancel) throws Exception {
+ final StreamTaskTestHarness<String> testHarness = new StreamTaskTestHarness<>(
+ SourceStreamTask::new,
+ BasicTypeInfo.STRING_TYPE_INFO);
+
+ CancelLockingSource.reset();
+ testHarness
+ .setupOperatorChain(
+ new OperatorID(),
+ new StreamSource<>(new CancelLockingSource(throwInCancel)))
+ .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) {
+ checkState(throwInCancel);
+ }
+
+ try {
+ testHarness.waitForTaskCompletion();
+ } catch (Throwable t) {
+ if (!ExceptionUtils.findThrowable(t, CancelTaskException.class).isPresent()) {
+ throw t;
+ }
+ }
+ }
+
+ /**
+ * A source that locks if cancellation attempts to cleanly shut down.
+ */
+ public static class CancelLockingSource implements SourceFunction<String> {
+ private static final long serialVersionUID = 8713065281092996042L;
+
+ private static CompletableFuture<Void> isRunning = new CompletableFuture<>();
+
+ private final boolean throwOnCancel;
+
+ private volatile boolean cancelled = false;
+
+ public CancelLockingSource(boolean throwOnCancel) {
+ this.throwOnCancel = throwOnCancel;
+ }
+
+ public static void reset() {
+ isRunning = new CompletableFuture<>();
+ }
+
+ public static void awaitRunning() throws ExecutionException, InterruptedException {
+ isRunning.get();
+ }
+
+ @Override
+ public void run(SourceContext<String> ctx) throws Exception {
+ synchronized (ctx.getCheckpointLock()) {
+ while (!cancelled) {
+ isRunning.complete(null);
+
+ if (throwOnCancel) {
+ Thread.sleep(1000000000);
+ }
+ else {
+ try {
+ //noinspection SleepWhileHoldingLock
+ Thread.sleep(1000000000);
+ } catch (InterruptedException ignored) {
+ }
+ }
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ if (throwOnCancel) {
+ throw new ExpectedTestException();
+ }
+ cancelled = true;
+ }
+ }
+
+ @Test
+ public void testInterruptedNotSwallowed() throws Exception {
+ final StreamTaskTestHarness<String> testHarness = new StreamTaskTestHarness<>(
+ SourceStreamTask::new,
+ BasicTypeInfo.STRING_TYPE_INFO);
+
+ CancelLockingSource.reset();
+ testHarness
+ .setupOperatorChain(
+ new OperatorID(),
+ new StreamSource<>(new InterruptedSource()))
+ .chain(
+ new OperatorID(),
+ new TestBoundedOneInputStreamOperator("Operator1"),
+ BasicTypeInfo.STRING_TYPE_INFO.createSerializer(new ExecutionConfig()))
+ .finish();
+
+ StreamConfig streamConfig = testHarness.getStreamConfig();
+ streamConfig.setTimeCharacteristic(TimeCharacteristic.ProcessingTime);
+
+ testHarness.invoke();
+ try {
+ testHarness.waitForTaskCompletion();
+ } catch (Exception e) {
+ if (!(e.getCause() instanceof InterruptedException)) {
+ throw e;
+ }
+ }
+ }
+
+ /**
+ * A source that locks if cancellation attempts to cleanly shut down.
+ */
+ public static class InterruptedSource implements SourceFunction<String> {
+ private static final long serialVersionUID = 8713065281092996042L;
+
+ private static CompletableFuture<Void> isRunning = new CompletableFuture<>();
+
+ public static void reset() {
+ isRunning = new CompletableFuture<>();
+ }
+
+ public static void awaitRunning() throws ExecutionException, InterruptedException {
+ isRunning.get();
+ }
+
+ @Override
+ public void run(SourceContext<String> ctx) throws Exception {
+ synchronized (ctx.getCheckpointLock()) {
+ isRunning.complete(null);
+ Thread.currentThread().interrupt();
+ throw new InterruptedException();
+ }
+ }
+
+ @Override
+ public void cancel() {
+ }
+ }
+
/**
* If finishing a task doesn't swallow exceptions this test would fail with an exception.
*/
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
index 1e64718..7719259 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskTest.java
@@ -315,26 +315,6 @@ public class StreamTaskTest extends TestLogger {
}
@Test
- public void testCancellationNotBlockedOnLock() throws Exception {
- syncLatch = new OneShotLatch();
-
- StreamConfig cfg = new StreamConfig(new Configuration());
- Task task = createTask(CancelLockingTask.class, cfg, new Configuration());
-
- // start the task and wait until it runs
- // execution state RUNNING is not enough, we need to wait until the stream task's run() method
- // is entered
- task.startTaskThread();
- syncLatch.await();
-
- // cancel the execution - this should lead to smooth shutdown
- task.cancelExecution();
- task.getExecutingThread().join();
-
- assertEquals(ExecutionState.CANCELED, task.getExecutionState());
- }
-
- @Test
public void testCancellationFailsWithBlockingLock() throws Exception {
syncLatch = new OneShotLatch();
@@ -1383,55 +1363,6 @@ public class StreamTaskTest extends TestLogger {
/**
* A task that locks if cancellation attempts to cleanly shut down.
*/
- public static class CancelLockingTask extends StreamTask<String, AbstractStreamOperator<String>> {
-
- private final OneShotLatch latch = new OneShotLatch();
-
- private LockHolder holder;
-
- public CancelLockingTask(Environment env) {
- super(env);
- }
-
- @Override
- protected void init() {}
-
- @Override
- protected void processInput(MailboxDefaultAction.Controller controller) throws Exception {
- holder = new LockHolder(getCheckpointLock(), latch);
- holder.start();
- latch.await();
-
- // we are at the point where cancelling can happen
- syncLatch.trigger();
-
- // just put this to sleep until it is interrupted
- try {
- Thread.sleep(100000000);
- } catch (InterruptedException ignored) {
- // restore interruption state
- Thread.currentThread().interrupt();
- }
- controller.allActionsCompleted();
- }
-
- @Override
- protected void cleanup() {
- holder.close();
- }
-
- @Override
- protected void cancelTask() {
- holder.cancel();
- // do not interrupt the lock holder here, to simulate spawned threads that
- // we cannot properly interrupt on cancellation
- }
-
- }
-
- /**
- * A task that locks if cancellation attempts to cleanly shut down.
- */
public static class CancelFailingTask extends StreamTask<String, AbstractStreamOperator<String>> {
public CancelFailingTask(Environment env) {
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java
index a63b102..efd70d9 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/MailboxExecutorImplTest.java
@@ -37,7 +37,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
-import static org.hamcrest.CoreMatchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
@@ -152,18 +151,6 @@ public class MailboxExecutorImplTest {
Assert.assertEquals(Thread.currentThread(), testRunnable.wasExecutedBy());
}
- @Test
- public void testPrettyExceptionMessage() {
- final String description = "Pretty command description";
- mailboxExecutor.execute(
- () -> {
- throw new RuntimeException("Some random exception");
- },
- description);
- expectedException.expectMessage(containsString(description));
- mailboxExecutor.tryYield();
- }
-
/**
* Test {@link Runnable} that tracks execution.
*/
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxProcessorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxProcessorTest.java
index ec14705..9740907 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxProcessorTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/mailbox/TaskMailboxProcessorTest.java
@@ -33,9 +33,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
-import static org.hamcrest.CoreMatchers.containsString;
-import static org.junit.internal.matchers.ThrowableMessageMatcher.hasMessage;
-
/**
* Unit tests for {@link MailboxProcessor}.
*/
@@ -59,7 +56,7 @@ public class TaskMailboxProcessorTest {
@Test
public void testSubmittingRunnableWithException() throws Exception {
- expectedException.expectCause(hasMessage(containsString("Expected")));
+ expectedException.expectMessage("Expected");
try (MailboxProcessor mailboxProcessor = new MailboxProcessor(controller -> {})) {
final Thread submitThread = new Thread(() -> {
mailboxProcessor.getMainMailboxExecutor().execute(