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 2021/02/11 18:34:06 UTC
[flink] branch release-1.11 updated (30351d6 -> 9946891)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 30351d6 [FLINK-21138] Explicit Classloader in QueryableStateClient
new b4e3b49 [FLINK-21274][runtime] Change the ClusterEntrypoint.runClusterEntrypoint to wait on the result of clusterEntrypoint.getTerminationFuture().get() and do the System.exit outside of the future callback
new 9946891 [FLINK-21274] Block main thread when running the TaskManagerRunner
The 2 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:
.../taskmanager/KubernetesTaskExecutorRunner.java | 2 +-
.../mesos/entrypoint/MesosTaskExecutorRunner.java | 2 +-
.../runtime/entrypoint/ClusterEntrypoint.java | 35 +++---
.../runtime/taskexecutor/TaskManagerRunner.java | 136 ++++++++++++++-------
.../taskexecutor/TaskManagerRunnerTest.java | 35 ++----
...tractTaskManagerProcessFailureRecoveryTest.java | 2 +-
.../apache/flink/yarn/YarnTaskExecutorRunner.java | 23 ++--
7 files changed, 137 insertions(+), 98 deletions(-)
[flink] 01/02: [FLINK-21274][runtime] Change the
ClusterEntrypoint.runClusterEntrypoint to wait on the result of
clusterEntrypoint.getTerminationFuture().get() and do the System.exit
outside of the future callback
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b4e3b498ec47deb383b83de87ac00b7707b26314
Author: wjc <wj...@foxmail.com>
AuthorDate: Tue Feb 9 11:07:51 2021 +0800
[FLINK-21274][runtime] Change the ClusterEntrypoint.runClusterEntrypoint to wait on the result of clusterEntrypoint.getTerminationFuture().get() and do the System.exit outside of the future callback
---
.../runtime/entrypoint/ClusterEntrypoint.java | 35 ++++++++++------------
1 file changed, 16 insertions(+), 19 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
index dd5e2a9..a824b52 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java
@@ -566,25 +566,22 @@ public abstract class ClusterEntrypoint implements AutoCloseableAsync, FatalErro
System.exit(STARTUP_FAILURE_RETURN_CODE);
}
- clusterEntrypoint
- .getTerminationFuture()
- .whenComplete(
- (applicationStatus, throwable) -> {
- final int returnCode;
-
- if (throwable != null) {
- returnCode = RUNTIME_FAILURE_RETURN_CODE;
- } else {
- returnCode = applicationStatus.processExitCode();
- }
-
- LOG.info(
- "Terminating cluster entrypoint process {} with exit code {}.",
- clusterEntrypointName,
- returnCode,
- throwable);
- System.exit(returnCode);
- });
+ int returnCode;
+ Throwable throwable = null;
+
+ try {
+ returnCode = clusterEntrypoint.getTerminationFuture().get().processExitCode();
+ } catch (Throwable e) {
+ throwable = ExceptionUtils.stripExecutionException(e);
+ returnCode = RUNTIME_FAILURE_RETURN_CODE;
+ }
+
+ LOG.info(
+ "Terminating cluster entrypoint process {} with exit code {}.",
+ clusterEntrypointName,
+ returnCode,
+ throwable);
+ System.exit(returnCode);
}
/** Execution mode of the {@link MiniDispatcher}. */
[flink] 02/02: [FLINK-21274] Block main thread when running the
TaskManagerRunner
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 9946891bb3d6568ae29af6d8c23ccb39bfbfba22
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Tue Feb 9 16:22:06 2021 +0100
[FLINK-21274] Block main thread when running the TaskManagerRunner
In order to ensure that the TaskManager properly shuts down we need to let
the main thread block on the execution of the TaskManager. This will ensure
that there is always a non-daemon thread as long as the TaskManager runs.
This closes #14915.
---
.../taskmanager/KubernetesTaskExecutorRunner.java | 2 +-
.../mesos/entrypoint/MesosTaskExecutorRunner.java | 2 +-
.../runtime/taskexecutor/TaskManagerRunner.java | 136 ++++++++++++++-------
.../taskexecutor/TaskManagerRunnerTest.java | 35 ++----
...tractTaskManagerProcessFailureRecoveryTest.java | 2 +-
.../apache/flink/yarn/YarnTaskExecutorRunner.java | 23 ++--
6 files changed, 121 insertions(+), 79 deletions(-)
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/taskmanager/KubernetesTaskExecutorRunner.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/taskmanager/KubernetesTaskExecutorRunner.java
index c0f40ba..ef412c8 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/taskmanager/KubernetesTaskExecutorRunner.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/taskmanager/KubernetesTaskExecutorRunner.java
@@ -43,6 +43,6 @@ public class KubernetesTaskExecutorRunner {
Preconditions.checkArgument(
resourceID != null, "Pod name variable %s not set", Constants.ENV_FLINK_POD_NAME);
- TaskManagerRunner.runTaskManagerSecurely(args, new ResourceID(resourceID));
+ TaskManagerRunner.runTaskManagerProcessSecurely(args, new ResourceID(resourceID));
}
}
diff --git a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
index 5811679..28ba83f 100644
--- a/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
+++ b/flink-mesos/src/main/java/org/apache/flink/mesos/entrypoint/MesosTaskExecutorRunner.java
@@ -81,6 +81,6 @@ public class MesosTaskExecutorRunner {
final ResourceID resourceId = new ResourceID(containerID);
LOG.info("ResourceID assigned for this container: {}", resourceId);
- TaskManagerRunner.runTaskManagerSecurely(configuration, resourceId);
+ TaskManagerRunner.runTaskManagerProcessSecurely(configuration, resourceId);
}
}
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 05e578d..a44ba84 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
@@ -74,6 +74,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -86,15 +87,14 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* constructs the related components (network, I/O manager, memory manager, RPC service, HA service)
* and starts them.
*/
-public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync {
+public class TaskManagerRunner implements FatalErrorHandler {
private static final Logger LOG = LoggerFactory.getLogger(TaskManagerRunner.class);
private static final long FATAL_ERROR_SHUTDOWN_TIMEOUT_MS = 10000L;
- private static final int STARTUP_FAILURE_RETURN_CODE = 1;
-
- public static final int RUNTIME_FAILURE_RETURN_CODE = 2;
+ private static final int SUCCESS_EXIT_CODE = 0;
+ private static final int FAILURE_EXIT_CODE = 1;
private final Object lock = new Object();
@@ -117,7 +117,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
private final TaskExecutorService taskExecutorService;
- private final CompletableFuture<Void> terminationFuture;
+ private final CompletableFuture<Result> terminationFuture;
private boolean shutdown;
@@ -183,7 +183,8 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
this.shutdown = false;
handleUnexpectedTaskExecutorServiceTermination();
- MemoryLogger.startIfConfigured(LOG, configuration, terminationFuture);
+ MemoryLogger.startIfConfigured(
+ LOG, configuration, terminationFuture.thenAccept(ignored -> {}));
}
private void handleUnexpectedTaskExecutorServiceTermination() {
@@ -210,8 +211,19 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
taskExecutorService.start();
}
- @Override
- public CompletableFuture<Void> closeAsync() {
+ public void close() throws Exception {
+ try {
+ closeAsync().get();
+ } catch (ExecutionException e) {
+ ExceptionUtils.rethrowException(ExceptionUtils.stripExecutionException(e));
+ }
+ }
+
+ public CompletableFuture<Result> closeAsync() {
+ return closeAsync(Result.SUCCESS);
+ }
+
+ private CompletableFuture<Result> closeAsync(Result terminationResult) {
synchronized (lock) {
if (!shutdown) {
shutdown = true;
@@ -228,7 +240,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
if (throwable != null) {
terminationFuture.completeExceptionally(throwable);
} else {
- terminationFuture.complete(null);
+ terminationFuture.complete(terminationResult);
}
});
}
@@ -275,7 +287,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
}
// export the termination future for caller to know it is terminated
- public CompletableFuture<Void> getTerminationFuture() {
+ public CompletableFuture<Result> getTerminationFuture() {
return terminationFuture;
}
@@ -299,20 +311,15 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
&& !ExceptionUtils.isMetaspaceOutOfMemoryError(enrichedException)) {
terminateJVM();
} else {
- closeAsync();
+ closeAsync(Result.FAILURE);
FutureUtils.orTimeout(
terminationFuture, FATAL_ERROR_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS);
-
- terminationFuture.whenComplete(
- (Void ignored, Throwable throwable) -> {
- terminateJVM();
- });
}
}
private void terminateJVM() {
- System.exit(RUNTIME_FAILURE_RETURN_CODE);
+ System.exit(FAILURE_EXIT_CODE);
}
// --------------------------------------------------------------------------------------------
@@ -333,7 +340,7 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
LOG.info("Cannot determine the maximum number of open file descriptors");
}
- runTaskManagerSecurely(args, ResourceID.generate());
+ runTaskManagerProcessSecurely(args, ResourceID.generate());
}
public static Configuration loadConfiguration(String[] args) throws FlinkParseException {
@@ -341,46 +348,74 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
args, TaskManagerRunner.class.getSimpleName());
}
- public static void runTaskManager(
+ public static int runTaskManager(
Configuration configuration, ResourceID resourceId, PluginManager pluginManager)
throws Exception {
- final TaskManagerRunner taskManagerRunner =
- new TaskManagerRunner(
- configuration,
- resourceId,
- pluginManager,
- TaskManagerRunner::createTaskExecutorService);
+ final TaskManagerRunner taskManagerRunner;
- taskManagerRunner.start();
- }
+ try {
+ taskManagerRunner =
+ new TaskManagerRunner(
+ configuration,
+ resourceId,
+ pluginManager,
+ TaskManagerRunner::createTaskExecutorService);
+ taskManagerRunner.start();
+ } catch (Exception exception) {
+ throw new FlinkException("Failed to start the TaskManagerRunner.", exception);
+ }
- public static void runTaskManagerSecurely(String[] args, ResourceID resourceID) {
try {
- Configuration configuration = loadConfiguration(args);
- runTaskManagerSecurely(configuration, resourceID);
+ return taskManagerRunner.getTerminationFuture().get().getExitCode();
} catch (Throwable t) {
- final Throwable strippedThrowable =
- ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
- LOG.error("TaskManager initialization failed.", strippedThrowable);
- System.exit(STARTUP_FAILURE_RETURN_CODE);
+ throw new FlinkException(
+ "Unexpected failure during runtime of TaskManagerRunner.",
+ ExceptionUtils.stripExecutionException(t));
}
}
- public static void runTaskManagerSecurely(Configuration configuration, ResourceID resourceID)
- throws Exception {
+ public static void runTaskManagerProcessSecurely(String[] args, ResourceID resourceID) {
+ Configuration configuration = null;
+
+ try {
+ configuration = loadConfiguration(args);
+ } catch (FlinkParseException fpe) {
+ LOG.error("Could not load the configuration.", fpe);
+ System.exit(FAILURE_EXIT_CODE);
+ }
+
+ runTaskManagerProcessSecurely(checkNotNull(configuration), resourceID);
+ }
+
+ public static void runTaskManagerProcessSecurely(
+ Configuration configuration, ResourceID resourceID) {
replaceGracefulExitWithHaltIfConfigured(configuration);
final PluginManager pluginManager =
PluginUtils.createPluginManagerFromRootFolder(configuration);
FileSystem.initialize(configuration, pluginManager);
- SecurityUtils.install(new SecurityConfiguration(configuration));
+ int exitCode;
+ Throwable throwable = null;
- SecurityUtils.getInstalledContext()
- .runSecured(
- () -> {
- runTaskManager(configuration, resourceID, pluginManager);
- return null;
- });
+ try {
+ SecurityUtils.install(new SecurityConfiguration(configuration));
+
+ exitCode =
+ SecurityUtils.getInstalledContext()
+ .runSecured(
+ () -> runTaskManager(configuration, resourceID, pluginManager));
+ } catch (Throwable t) {
+ throwable = ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
+ exitCode = FAILURE_EXIT_CODE;
+ }
+
+ if (throwable != null) {
+ LOG.error("Terminating TaskManagerRunner with exit code {}.", exitCode, throwable);
+ } else {
+ LOG.info("Terminating TaskManagerRunner with exit code {}.", exitCode);
+ }
+
+ System.exit(exitCode);
}
// --------------------------------------------------------------------------------------------
@@ -581,4 +616,19 @@ public class TaskManagerRunner implements FatalErrorHandler, AutoCloseableAsync
CompletableFuture<Void> getTerminationFuture();
}
+
+ public enum Result {
+ SUCCESS(SUCCESS_EXIT_CODE),
+ FAILURE(FAILURE_EXIT_CODE);
+
+ private final int exitCode;
+
+ Result(int exitCode) {
+ this.exitCode = exitCode;
+ }
+
+ public int getExitCode() {
+ return exitCode;
+ }
+ }
}
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 6282f4d..c6e0adb 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
@@ -24,23 +24,18 @@ import org.apache.flink.configuration.TaskManagerOptions;
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;
import org.junit.After;
-import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
import javax.annotation.Nonnull;
-import java.time.Duration;
import java.util.concurrent.CompletableFuture;
-import static org.apache.flink.core.testutils.FlinkMatchers.willNotComplete;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
@@ -50,15 +45,8 @@ public class TaskManagerRunnerTest extends TestLogger {
@Rule public final Timeout timeout = Timeout.seconds(30);
- private SystemExitTrackingSecurityManager systemExitTrackingSecurityManager;
private TaskManagerRunner taskManagerRunner;
- @Before
- public void before() {
- systemExitTrackingSecurityManager = new SystemExitTrackingSecurityManager();
- System.setSecurityManager(systemExitTrackingSecurityManager);
- }
-
@After
public void after() throws Exception {
System.setSecurityManager(null);
@@ -76,8 +64,9 @@ public class TaskManagerRunnerTest extends TestLogger {
taskManagerRunner.onFatalError(new RuntimeException());
- Integer statusCode = systemExitTrackingSecurityManager.getSystemExitFuture().get();
- assertThat(statusCode, is(equalTo(TaskManagerRunner.RUNTIME_FAILURE_RETURN_CODE)));
+ assertThat(
+ taskManagerRunner.getTerminationFuture().join(),
+ is(equalTo(TaskManagerRunner.Result.FAILURE)));
}
@Test
@@ -87,8 +76,9 @@ public class TaskManagerRunnerTest extends TestLogger {
TaskManagerOptions.REGISTRATION_TIMEOUT, TimeUtils.parseDuration("10 ms"));
taskManagerRunner = createTaskManagerRunner(configuration);
- Integer statusCode = systemExitTrackingSecurityManager.getSystemExitFuture().get();
- assertThat(statusCode, is(equalTo(TaskManagerRunner.RUNTIME_FAILURE_RETURN_CODE)));
+ assertThat(
+ taskManagerRunner.getTerminationFuture().join(),
+ is(equalTo(TaskManagerRunner.Result.FAILURE)));
}
@Test
@@ -103,10 +93,11 @@ public class TaskManagerRunnerTest extends TestLogger {
createConfiguration(),
createTaskExecutorServiceFactory(taskExecutorService));
- terminationFuture.completeExceptionally(new FlinkException("Test exception."));
+ terminationFuture.complete(null);
- Integer statusCode = systemExitTrackingSecurityManager.getSystemExitFuture().get();
- assertThat(statusCode, is(equalTo(TaskManagerRunner.RUNTIME_FAILURE_RETURN_CODE)));
+ assertThat(
+ taskManagerRunner.getTerminationFuture().join(),
+ is(equalTo(TaskManagerRunner.Result.FAILURE)));
}
@Test
@@ -125,11 +116,11 @@ public class TaskManagerRunnerTest extends TestLogger {
taskManagerRunner.closeAsync();
- terminationFuture.completeExceptionally(new FlinkException("Test exception."));
+ terminationFuture.complete(null);
assertThat(
- systemExitTrackingSecurityManager.getSystemExitFuture(),
- willNotComplete(Duration.ofMillis(10L)));
+ taskManagerRunner.getTerminationFuture().join(),
+ is(equalTo(TaskManagerRunner.Result.SUCCESS)));
}
@Nonnull
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java
index 1abb02f..4fcb56f 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/AbstractTaskManagerProcessFailureRecoveryTest.java
@@ -337,7 +337,7 @@ public abstract class AbstractTaskManagerProcessFailureRecoveryTest extends Test
TaskManagerRunner.runTaskManager(cfg, ResourceID.generate(), pluginManager);
} catch (Throwable t) {
- LOG.error("Failed to start TaskManager process", t);
+ LOG.error("Failed to run the TaskManager process", t);
System.exit(1);
}
}
diff --git a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
index 7d87e2b..681f670 100644
--- a/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
+++ b/flink-yarn/src/main/java/org/apache/flink/yarn/YarnTaskExecutorRunner.java
@@ -29,7 +29,6 @@ import org.apache.flink.runtime.taskexecutor.TaskManagerRunner;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.runtime.util.JvmShutdownSafeguard;
import org.apache.flink.runtime.util.SignalHandler;
-import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
import org.apache.hadoop.security.UserGroupInformation;
@@ -38,7 +37,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.lang.reflect.UndeclaredThrowableException;
import java.util.Map;
/** This class is the executable entry point for running a TaskExecutor in a YARN container. */
@@ -71,35 +69,38 @@ public class YarnTaskExecutorRunner {
/**
* The instance entry point for the YARN task executor. Obtains user group information and calls
- * the main work method {@link TaskManagerRunner#runTaskManager(Configuration, ResourceID)} as a
- * privileged action.
+ * the main work method {@link TaskManagerRunner#runTaskManagerProcessSecurely(Configuration,
+ * ResourceID)} as a privileged action.
*
* @param args The command line arguments.
*/
private static void runTaskManagerSecurely(String[] args) {
+ Configuration configuration = null;
+ String containerId = null;
+
try {
LOG.debug("All environment variables: {}", ENV);
final String currDir = ENV.get(Environment.PWD.key());
LOG.info("Current working Directory: {}", currDir);
- final Configuration configuration = TaskManagerRunner.loadConfiguration(args);
+ configuration = TaskManagerRunner.loadConfiguration(args);
setupAndModifyConfiguration(configuration, currDir, ENV);
- final String containerId = ENV.get(YarnResourceManager.ENV_FLINK_CONTAINER_ID);
+ containerId = ENV.get(YarnResourceManager.ENV_FLINK_CONTAINER_ID);
Preconditions.checkArgument(
containerId != null,
"ContainerId variable %s not set",
YarnResourceManager.ENV_FLINK_CONTAINER_ID);
- TaskManagerRunner.runTaskManagerSecurely(configuration, new ResourceID(containerId));
} catch (Throwable t) {
- final Throwable strippedThrowable =
- ExceptionUtils.stripException(t, UndeclaredThrowableException.class);
- // make sure that everything whatever ends up in the log
- LOG.error("YARN TaskManager initialization failed.", strippedThrowable);
+ LOG.error("YARN TaskManager initialization failed.", t);
System.exit(INIT_ERROR_EXIT_CODE);
}
+
+ TaskManagerRunner.runTaskManagerProcessSecurely(
+ Preconditions.checkNotNull(configuration),
+ new ResourceID(Preconditions.checkNotNull(containerId)));
}
@VisibleForTesting