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 2021/10/08 08:01:15 UTC
[flink] 01/10: [FLINK-24182][task] Do not interrupt tasks/operators
that are already closing
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 53592f88d4b696da7e511ba6d23835be7db39aeb
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Fri Sep 10 15:16:27 2021 +0200
[FLINK-24182][task] Do not interrupt tasks/operators that are already closing
Otherwise if we interrupt operator that is doing a blocking call to release
some resources in closing, we can cause those resources to leak. A good example
is Kafka sink, which wants to abort transactions in close() method.
---
.../flink/streaming/runtime/tasks/StreamTask.java | 14 +++++---
.../runtime/tasks/StreamTaskCancellationTest.java | 39 ++++++++++++++++++++++
2 files changed, 48 insertions(+), 5 deletions(-)
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 870bd59..093a03a 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
@@ -874,6 +874,8 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
LOG.debug("All pending checkpoints are finished");
}
+ disableInterruptOnCancel();
+
// make an attempt to dispose the operators such that failures in the dispose call
// still let the computation fail
closeAllOperators();
@@ -902,11 +904,6 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
// clean up everything we initialized
isRunning = false;
- // Now that we are outside the user code, we do not want to be interrupted further
- // upon cancellation. The shutdown logic below needs to make sure it does not issue calls
- // that block and stall shutdown.
- // Additionally, the cancellation watch dog will issue a hard-cancel (kill the TaskManager
- // process) as a backup in case some shutdown procedure blocks outside our control.
disableInterruptOnCancel();
// clear any previously issued interrupt for a more graceful shutdown
@@ -1752,6 +1749,13 @@ public abstract class StreamTask<OUT, OP extends StreamOperator<OUT>>
return true;
}
+ /**
+ * While we are outside the user code, we do not want to be interrupted further upon
+ * cancellation. The shutdown logic below needs to make sure it does not issue calls that block
+ * and stall shutdown. Additionally, the cancellation watch dog will issue a hard-cancel (kill
+ * the TaskManager process) as a backup in case some shutdown procedure blocks outside our
+ * control.
+ */
private void disableInterruptOnCancel() {
synchronized (shouldInterruptOnCancelLock) {
shouldInterruptOnCancel = false;
diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationTest.java
index 6d1b60d..cdbc51a 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/tasks/StreamTaskCancellationTest.java
@@ -35,8 +35,10 @@ import org.apache.flink.runtime.taskmanager.TestTaskBuilder;
import org.apache.flink.runtime.util.NettyShuffleDescriptorBuilder;
import org.apache.flink.streaming.api.graph.StreamConfig;
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
+import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.io.DataInputStatus;
import org.apache.flink.streaming.runtime.io.StreamInputProcessor;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.streaming.runtime.tasks.mailbox.MailboxDefaultAction;
import org.apache.flink.util.TestLogger;
@@ -46,8 +48,10 @@ import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicBoolean;
import static java.util.Collections.singletonList;
+import static org.apache.flink.api.common.typeinfo.BasicTypeInfo.STRING_TYPE_INFO;
import static org.apache.flink.streaming.runtime.tasks.StreamTaskTest.createTask;
import static org.apache.flink.util.Preconditions.checkState;
import static org.junit.Assert.assertEquals;
@@ -57,6 +61,41 @@ import static org.junit.Assert.assertFalse;
public class StreamTaskCancellationTest extends TestLogger {
@Test
+ public void testDoNotInterruptWhileClosing() throws Exception {
+ TestInterruptInCloseOperator testOperator = new TestInterruptInCloseOperator();
+ try (StreamTaskMailboxTestHarness<String> harness =
+ new StreamTaskMailboxTestHarnessBuilder<>(OneInputStreamTask::new, STRING_TYPE_INFO)
+ .addInput(STRING_TYPE_INFO)
+ .setupOutputForSingletonOperatorChain(testOperator)
+ .build()) {}
+ }
+
+ private static class TestInterruptInCloseOperator extends AbstractStreamOperator<String>
+ implements OneInputStreamOperator<String, String> {
+ @Override
+ public void close() throws Exception {
+ super.close();
+
+ AtomicBoolean running = new AtomicBoolean(true);
+ Thread thread =
+ new Thread(
+ () -> {
+ while (running.get()) {}
+ });
+ thread.start();
+ try {
+ getContainingTask().maybeInterruptOnCancel(thread, null, null);
+ assertFalse(thread.isInterrupted());
+ } finally {
+ running.set(false);
+ }
+ }
+
+ @Override
+ public void processElement(StreamRecord<String> element) throws Exception {}
+ }
+
+ @Test
public void testCancellationWaitsForActiveTimers() throws Exception {
StreamTaskWithBlockingTimer.reset();
ResultPartitionDeploymentDescriptor descriptor =