You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2020/10/20 21:03:14 UTC
[flink] 09/09: [FLINK-19022] Fail TaskManagerRunner fatally if
TaskExecutor terminates unexpectedly
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 0ff7d7a6ddf28237797ff1c9ed6fc525a9245d90
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Oct 20 13:47:38 2020 +0200
[FLINK-19022] Fail TaskManagerRunner fatally if TaskExecutor terminates unexpectedly
An unexpected termination of the TaskExecutor indicates that the system has reached an
invalid state. Therefore, the runner will fail fatally if this happens.
This closes #13319.
---
.../taskexecutor/TaskExecutorToServiceAdapter.java | 52 ++++++++++++
.../runtime/taskexecutor/TaskManagerRunner.java | 77 ++++++++++++++++--
.../taskexecutor/TaskManagerRunnerTest.java | 62 ++++++++++++++-
.../taskexecutor/TestingTaskExecutorService.java | 92 ++++++++++++++++++++++
.../JobManagerHAProcessFailureRecoveryITCase.java | 2 +-
5 files changed, 277 insertions(+), 8 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToServiceAdapter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToServiceAdapter.java
new file mode 100644
index 0000000..14512e3
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToServiceAdapter.java
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskexecutor;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Simple adapter for {@link TaskExecutor} to adapt to {@link TaskManagerRunner.TaskExecutorService}.
+ */
+public class TaskExecutorToServiceAdapter implements TaskManagerRunner.TaskExecutorService {
+
+ private final TaskExecutor taskExecutor;
+
+ private TaskExecutorToServiceAdapter(TaskExecutor taskExecutor) {
+ this.taskExecutor = taskExecutor;
+ }
+
+ @Override
+ public void start() {
+ taskExecutor.start();
+ }
+
+ @Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return taskExecutor.getTerminationFuture();
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ return taskExecutor.closeAsync();
+ }
+
+ public static TaskExecutorToServiceAdapter createFor(TaskExecutor taskExecutor) {
+ return new TaskExecutorToServiceAdapter(taskExecutor);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
index b296baf..1c5ce10 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
@@ -67,6 +67,7 @@ import org.apache.flink.util.AbstractID;
import org.apache.flink.util.AutoCloseableAsync;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TaskManagerExceptionUtils;
@@ -121,13 +122,16 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
/** Executor used to run future callbacks. */
private final ExecutorService executor;
- private final TaskExecutor taskManager;
+ private final TaskExecutorService taskExecutorService;
private final CompletableFuture<Void> terminationFuture;
private boolean shutdown;
- public TaskManagerRunner(Configuration configuration, PluginManager pluginManager) throws Exception {
+ public TaskManagerRunner(
+ Configuration configuration,
+ PluginManager pluginManager,
+ TaskExecutorServiceFactory taskExecutorServiceFactory) throws Exception {
this.configuration = checkNotNull(configuration);
timeout = AkkaUtils.getTimeoutAsTime(configuration);
@@ -165,7 +169,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
ExternalResourceUtils.getExternalResourceAmountMap(configuration),
ExternalResourceUtils.externalResourceDriversFromConfig(configuration, pluginManager));
- taskManager = startTaskManager(
+ taskExecutorService = taskExecutorServiceFactory.createTaskExecutor(
this.configuration,
this.resourceId,
rpcService,
@@ -179,16 +183,27 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
this.terminationFuture = new CompletableFuture<>();
this.shutdown = false;
+ handleUnexpectedTaskExecutorServiceTermination();
MemoryLogger.startIfConfigured(LOG, configuration, terminationFuture);
}
+ private void handleUnexpectedTaskExecutorServiceTermination() {
+ taskExecutorService.getTerminationFuture().whenComplete((unused, throwable) -> {
+ synchronized (lock) {
+ if (!shutdown) {
+ onFatalError(new FlinkException("Unexpected termination of the TaskExecutor.", throwable));
+ }
+ }
+ });
+ }
+
// --------------------------------------------------------------------------------------------
// Lifecycle management
// --------------------------------------------------------------------------------------------
public void start() throws Exception {
- taskManager.start();
+ taskExecutorService.start();
}
@Override
@@ -197,7 +212,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
if (!shutdown) {
shutdown = true;
- final CompletableFuture<Void> taskManagerTerminationFuture = taskManager.closeAsync();
+ final CompletableFuture<Void> taskManagerTerminationFuture = taskExecutorService.closeAsync();
final CompletableFuture<Void> serviceTerminationFuture = FutureUtils.composeAfterwards(
taskManagerTerminationFuture,
@@ -317,7 +332,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
}
public static void runTaskManager(Configuration configuration, PluginManager pluginManager) throws Exception {
- final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager);
+ final TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager, TaskManagerRunner::createTaskExecutorService);
taskManagerRunner.start();
}
@@ -350,6 +365,33 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
// Static utilities
// --------------------------------------------------------------------------------------------
+ public static TaskExecutorService createTaskExecutorService(
+ Configuration configuration,
+ ResourceID resourceID,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
+ boolean localCommunicationOnly,
+ ExternalResourceInfoProvider externalResourceInfoProvider,
+ FatalErrorHandler fatalErrorHandler) throws Exception {
+
+ final TaskExecutor taskExecutor = startTaskManager(
+ configuration,
+ resourceID,
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ metricRegistry,
+ blobCacheService,
+ localCommunicationOnly,
+ externalResourceInfoProvider,
+ fatalErrorHandler);
+
+ return TaskExecutorToServiceAdapter.createFor(taskExecutor);
+ }
+
public static TaskExecutor startTaskManager(
Configuration configuration,
ResourceID resourceID,
@@ -489,4 +531,27 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
: rpcAddress + ":" + rpcPort + "-" + new AbstractID().toString().substring(0, 6)),
config.getString(TaskManagerOptionsInternal.TASK_MANAGER_RESOURCE_ID_METADATA, ""));
}
+
+ /**
+ * Factory for {@link TaskExecutor}.
+ */
+ public interface TaskExecutorServiceFactory {
+ TaskExecutorService createTaskExecutor(
+ Configuration configuration,
+ ResourceID resourceID,
+ RpcService rpcService,
+ HighAvailabilityServices highAvailabilityServices,
+ HeartbeatServices heartbeatServices,
+ MetricRegistry metricRegistry,
+ BlobCacheService blobCacheService,
+ boolean localCommunicationOnly,
+ ExternalResourceInfoProvider externalResourceInfoProvider,
+ FatalErrorHandler fatalErrorHandler) throws Exception;
+ }
+
+ public interface TaskExecutorService extends AutoCloseableAsync {
+ void start();
+
+ CompletableFuture<Void> getTerminationFuture();
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java
index ca8f91d..667b88c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunnerTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.core.plugin.PluginManager;
import org.apache.flink.core.plugin.PluginUtils;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.testutils.SystemExitTrackingSecurityManager;
+import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
import org.apache.flink.util.TimeUtils;
@@ -35,8 +36,13 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
+import javax.annotation.Nonnull;
+
import java.net.InetAddress;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
@@ -144,6 +150,56 @@ public class TaskManagerRunnerTest extends TestLogger {
assertThat(taskManagerResourceID.getResourceIdString(), containsString(InetAddress.getLocalHost().getHostName()));
}
+ @Test
+ public void testUnexpectedTaskManagerTerminationFailsRunnerFatally() throws Exception {
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ final TestingTaskExecutorService taskExecutorService = TestingTaskExecutorService.newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .build();
+ final TaskManagerRunner taskManagerRunner = createTaskManagerRunner(
+ createConfiguration(),
+ createTaskExecutorServiceFactory(taskExecutorService));
+
+ terminationFuture.completeExceptionally(new FlinkException("Test exception."));
+
+ Integer statusCode = systemExitTrackingSecurityManager.getSystemExitFuture().get();
+ assertThat(statusCode, is(equalTo(TaskManagerRunner.RUNTIME_FAILURE_RETURN_CODE)));
+ }
+
+ @Test
+ public void testUnexpectedTaskManagerTerminationAfterRunnerCloseWillBeIgnored() throws Exception {
+ final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ final TestingTaskExecutorService taskExecutorService = TestingTaskExecutorService.newBuilder()
+ .setTerminationFuture(terminationFuture)
+ .withManualTerminationFutureCompletion()
+ .build();
+ final TaskManagerRunner taskManagerRunner = createTaskManagerRunner(
+ createConfiguration(),
+ createTaskExecutorServiceFactory(taskExecutorService));
+
+ taskManagerRunner.closeAsync();
+
+ terminationFuture.completeExceptionally(new FlinkException("Test exception."));
+
+ assertThat(systemExitTrackingSecurityManager.getSystemExitFuture(), willNotComplete(Duration.ofMillis(10L)));
+ }
+
+ @Nonnull
+ private TaskManagerRunner.TaskExecutorServiceFactory createTaskExecutorServiceFactory(
+ TestingTaskExecutorService taskExecutorService) {
+ return (
+ configuration,
+ resourceID,
+ rpcService,
+ highAvailabilityServices,
+ heartbeatServices,
+ metricRegistry,
+ blobCacheService,
+ localCommunicationOnly,
+ externalResourceInfoProvider,
+ fatalErrorHandler) -> taskExecutorService;
+ }
+
private static Configuration createConfiguration() {
final Configuration configuration = new Configuration();
configuration.setString(JobManagerOptions.ADDRESS, "localhost");
@@ -152,8 +208,12 @@ public class TaskManagerRunnerTest extends TestLogger {
}
private static TaskManagerRunner createTaskManagerRunner(final Configuration configuration) throws Exception {
+ return createTaskManagerRunner(configuration, TaskManagerRunner::createTaskExecutorService);
+ }
+
+ private static TaskManagerRunner createTaskManagerRunner(final Configuration configuration, TaskManagerRunner.TaskExecutorServiceFactory taskExecutorServiceFactory) throws Exception {
final PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(configuration);
- TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager);
+ TaskManagerRunner taskManagerRunner = new TaskManagerRunner(configuration, pluginManager, taskExecutorServiceFactory);
taskManagerRunner.start();
return taskManagerRunner;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorService.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorService.java
new file mode 100644
index 0000000..b9bd56c
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorService.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.taskexecutor;
+
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Testing implementation of {@link TaskManagerRunner.TaskExecutorService}.
+ */
+public class TestingTaskExecutorService implements TaskManagerRunner.TaskExecutorService {
+ private final Runnable startRunnable;
+ private final CompletableFuture<Void> terminationFuture;
+ private final boolean completeTerminationFutureOnClose;
+
+ private TestingTaskExecutorService(
+ Runnable startRunnable,
+ CompletableFuture<Void> terminationFuture,
+ boolean completeTerminationFutureOnClose) {
+ this.startRunnable = startRunnable;
+ this.terminationFuture = terminationFuture;
+ this.completeTerminationFutureOnClose = completeTerminationFutureOnClose;
+ }
+
+ @Override
+ public void start() {
+ startRunnable.run();
+ }
+
+ @Override
+ public CompletableFuture<Void> getTerminationFuture() {
+ return terminationFuture;
+ }
+
+ @Override
+ public CompletableFuture<Void> closeAsync() {
+ if (completeTerminationFutureOnClose) {
+ terminationFuture.complete(null);
+ }
+ return terminationFuture;
+ }
+
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ /**
+ * Builder for {@link TestingTaskExecutorService}.
+ */
+ public static final class Builder {
+ private Runnable startRunnable = () -> {};
+ private CompletableFuture<Void> terminationFuture = new CompletableFuture<>();
+ private boolean completeTerminationFutureOnClose = true;
+
+ public Builder setStartRunnable(Runnable startRunnable) {
+ this.startRunnable = startRunnable;
+ return this;
+ }
+
+ public Builder setTerminationFuture(CompletableFuture<Void> terminationFuture) {
+ this.terminationFuture = terminationFuture;
+ return this;
+ }
+
+ public Builder withManualTerminationFutureCompletion() {
+ completeTerminationFutureOnClose = false;
+ return this;
+ }
+
+ TestingTaskExecutorService build() {
+ return new TestingTaskExecutorService(
+ startRunnable,
+ terminationFuture,
+ completeTerminationFutureOnClose);
+ }
+ }
+}
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
index cdbb7d8..a4bd8fc 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/JobManagerHAProcessFailureRecoveryITCase.java
@@ -273,7 +273,7 @@ public class JobManagerHAProcessFailureRecoveryITCase extends TestLogger {
final PluginManager pluginManager = PluginUtils.createPluginManagerFromRootFolder(config);
// Start the task manager process
for (int i = 0; i < numberOfTaskManagers; i++) {
- taskManagerRunners[i] = new TaskManagerRunner(config, pluginManager);
+ taskManagerRunners[i] = new TaskManagerRunner(config, pluginManager, TaskManagerRunner::createTaskExecutorService);
taskManagerRunners[i].start();
}