KUBERNETES_PERSISTENT_VOLUME_CLAIM_READ_ONLY =
key("kubernetes.persistent-volume-claim-read-only")
.booleanType()
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecorator.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecorator.java
new file mode 100644
index 00000000000000..68562229b77f71
--- /dev/null
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecorator.java
@@ -0,0 +1,143 @@
+/*
+ * 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.kubernetes.kubeclient.decorators;
+
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.kubernetes.kubeclient.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.parameters.AbstractKubernetesParameters;
+
+import io.fabric8.kubernetes.api.model.Container;
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ExecActionBuilder;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+
+import java.time.Duration;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Sets {@code terminationGracePeriodSeconds} and a {@code preStop} lifecycle hook on a pod, gated
+ * by a termination-grace-period {@link ConfigOption} (see {@code
+ * KubernetesConfigOptions#TASK_MANAGER_TERMINATION_GRACE_PERIOD}). Only wired in for the
+ * TaskManager today; the {@link ConfigOption} is taken as a constructor argument so the same
+ * decorator can be reused for the JobManager once it has an equivalent {@code SIGUSR2} handler.
+ *
+ * The {@code preStop} hook sends {@code SIGUSR2} to the main container process and then sleeps
+ * for most of the remaining grace period before returning - Kubernetes only sends {@code SIGTERM}
+ * once {@code preStop} returns, so without that sleep the grace period configured here would never
+ * actually be available to already-running tasks; {@code SIGTERM} would follow the signal almost
+ * immediately. {@code TaskManagerRunner} interprets {@code SIGUSR2} as a request to prepare for
+ * graceful termination: it disconnects from the ResourceManager immediately instead of waiting to
+ * be timed out, so the ResourceManager frees this TaskExecutor's slots right away rather than after
+ * the heartbeat timeout elapses, while currently-running tasks keep running for the rest of the
+ * sleep. A {@code SIGUSR2} handler must exist on the receiving side before this decorator is wired
+ * in for a given pod type - Kubernetes' and the JVM's default disposition for an unhandled {@code
+ * SIGUSR2} is to terminate the process, which would make termination less graceful, not more.
+ *
+ *
Does nothing if the corresponding option is unset, and never overwrites a {@code
+ * terminationGracePeriodSeconds} or {@code preStop} hook already present in the user's pod
+ * template.
+ */
+public class TerminationGracePeriodDecorator extends AbstractKubernetesStepDecorator {
+
+ /**
+ * How much of the grace period to reserve for the JVM's own shutdown sequence to run after
+ * {@code SIGTERM}, once the {@code preStop} hook returns and Kubernetes sends it.
+ */
+ private static final long SAFETY_MARGIN_SECONDS = 5;
+
+ private final AbstractKubernetesParameters kubernetesParameters;
+ private final ConfigOption terminationGracePeriodOption;
+
+ public TerminationGracePeriodDecorator(
+ AbstractKubernetesParameters kubernetesParameters,
+ ConfigOption terminationGracePeriodOption) {
+ this.kubernetesParameters = checkNotNull(kubernetesParameters);
+ this.terminationGracePeriodOption = checkNotNull(terminationGracePeriodOption);
+ }
+
+ @Override
+ public FlinkPod decorateFlinkPod(FlinkPod flinkPod) {
+ final Configuration flinkConfig = kubernetesParameters.getFlinkConfiguration();
+ if (!flinkConfig.contains(terminationGracePeriodOption)) {
+ return flinkPod;
+ }
+ final Duration terminationGracePeriod = flinkConfig.get(terminationGracePeriodOption);
+
+ final PodBuilder podBuilder = new PodBuilder(flinkPod.getPodWithoutMainContainer());
+ if (flinkPod.getPodWithoutMainContainer().getSpec().getTerminationGracePeriodSeconds()
+ == null) {
+ podBuilder
+ .editOrNewSpec()
+ .withTerminationGracePeriodSeconds(terminationGracePeriod.getSeconds())
+ .endSpec();
+ } else {
+ logger.info(
+ "Not overwriting terminationGracePeriodSeconds already set by the pod "
+ + "template; '{}' has no effect on the grace period.",
+ terminationGracePeriodOption.key());
+ }
+
+ final Container mainContainer = flinkPod.getMainContainer();
+ final Container decoratedMainContainer;
+ if (mainContainer.getLifecycle() != null
+ && mainContainer.getLifecycle().getPreStop() != null) {
+ logger.info(
+ "Not overwriting the preStop hook already set by the pod template; '{}' has "
+ + "no effect on the container lifecycle.",
+ terminationGracePeriodOption.key());
+ decoratedMainContainer = mainContainer;
+ } else {
+ decoratedMainContainer =
+ new ContainerBuilder(mainContainer)
+ .editOrNewLifecycle()
+ .withNewPreStop()
+ .withExec(
+ new ExecActionBuilder()
+ .withCommand(
+ "sh",
+ "-c",
+ buildGracefulTerminationCommand(
+ terminationGracePeriod))
+ .build())
+ .endPreStop()
+ .endLifecycle()
+ .build();
+ }
+
+ return new FlinkPod.Builder(flinkPod)
+ .withPod(podBuilder.build())
+ .withMainContainer(decoratedMainContainer)
+ .build();
+ }
+
+ /**
+ * {@code 1} is the main container process because Flink's docker-entrypoint.sh execs it
+ * directly rather than running it as a child of the shell. The sleep keeps {@code preStop} from
+ * returning - and therefore Kubernetes from sending {@code SIGTERM} - until most of the grace
+ * period has elapsed, leaving {@link #SAFETY_MARGIN_SECONDS} for the JVM's own shutdown
+ * sequence to run afterwards.
+ */
+ private static String buildGracefulTerminationCommand(Duration terminationGracePeriod) {
+ final long sleepSeconds =
+ Math.max(0, terminationGracePeriod.getSeconds() - SAFETY_MARGIN_SECONDS);
+ return String.format("kill -USR2 1 2>/dev/null || true; sleep %d", sleepSeconds);
+ }
+}
diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java
index 05ed8c91cbf4df..3896c7b9683cac 100644
--- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java
+++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/factory/KubernetesTaskManagerFactory.java
@@ -29,6 +29,7 @@
import org.apache.flink.kubernetes.kubeclient.decorators.KubernetesStepDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.MountSecretsDecorator;
import org.apache.flink.kubernetes.kubeclient.decorators.PersistentVolumeClaimMountDecorator;
+import org.apache.flink.kubernetes.kubeclient.decorators.TerminationGracePeriodDecorator;
import org.apache.flink.kubernetes.kubeclient.parameters.KubernetesTaskManagerParameters;
import org.apache.flink.kubernetes.kubeclient.resources.KubernetesPod;
import org.apache.flink.util.Preconditions;
@@ -42,6 +43,7 @@
import static org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.KUBERNETES_HADOOP_CONF_MOUNT_DECORATOR_ENABLED;
import static org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.KUBERNETES_KERBEROS_MOUNT_DECORATOR_ENABLED;
+import static org.apache.flink.kubernetes.configuration.KubernetesConfigOptions.TASK_MANAGER_TERMINATION_GRACE_PERIOD;
/** Utility class for constructing the TaskManager Pod on the JobManager. */
public class KubernetesTaskManagerFactory {
@@ -58,7 +60,10 @@ public static KubernetesPod buildTaskManagerKubernetesPod(
new MountSecretsDecorator(kubernetesTaskManagerParameters),
new PersistentVolumeClaimMountDecorator(
kubernetesTaskManagerParameters),
- new CmdTaskManagerDecorator(kubernetesTaskManagerParameters)));
+ new CmdTaskManagerDecorator(kubernetesTaskManagerParameters),
+ new TerminationGracePeriodDecorator(
+ kubernetesTaskManagerParameters,
+ TASK_MANAGER_TERMINATION_GRACE_PERIOD)));
Configuration configuration = kubernetesTaskManagerParameters.getFlinkConfiguration();
if (configuration.get(KUBERNETES_HADOOP_CONF_MOUNT_DECORATOR_ENABLED)) {
diff --git a/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecoratorTest.java b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecoratorTest.java
new file mode 100644
index 00000000000000..9c9e6dd4a3ab32
--- /dev/null
+++ b/flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecoratorTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.kubernetes.kubeclient.decorators;
+
+import org.apache.flink.kubernetes.configuration.KubernetesConfigOptions;
+import org.apache.flink.kubernetes.kubeclient.FlinkPod;
+import org.apache.flink.kubernetes.kubeclient.KubernetesTaskManagerTestBase;
+
+import io.fabric8.kubernetes.api.model.ContainerBuilder;
+import io.fabric8.kubernetes.api.model.ExecActionBuilder;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for the {@link TerminationGracePeriodDecorator}. */
+class TerminationGracePeriodDecoratorTest extends KubernetesTaskManagerTestBase {
+
+ private static final Duration GRACE_PERIOD = Duration.ofSeconds(45);
+
+ private TerminationGracePeriodDecorator decorator;
+
+ @Override
+ public void onSetup() throws Exception {
+ super.onSetup();
+ this.decorator =
+ new TerminationGracePeriodDecorator(
+ kubernetesTaskManagerParameters,
+ KubernetesConfigOptions.TASK_MANAGER_TERMINATION_GRACE_PERIOD);
+ }
+
+ @Test
+ void testUnsetOptionLeavesPodUnchanged() {
+ final FlinkPod resultFlinkPod = decorator.decorateFlinkPod(baseFlinkPod);
+
+ assertThat(resultFlinkPod.getPodWithoutMainContainer())
+ .isEqualTo(baseFlinkPod.getPodWithoutMainContainer());
+ assertThat(resultFlinkPod.getMainContainer()).isEqualTo(baseFlinkPod.getMainContainer());
+ }
+
+ @Test
+ void testSetOptionAddsGracePeriodAndPreStopHook() {
+ flinkConfig.set(
+ KubernetesConfigOptions.TASK_MANAGER_TERMINATION_GRACE_PERIOD, GRACE_PERIOD);
+
+ final FlinkPod resultFlinkPod = decorator.decorateFlinkPod(baseFlinkPod);
+
+ assertThat(
+ resultFlinkPod
+ .getPodWithoutMainContainer()
+ .getSpec()
+ .getTerminationGracePeriodSeconds())
+ .isEqualTo(GRACE_PERIOD.getSeconds());
+ assertThat(resultFlinkPod.getMainContainer().getLifecycle())
+ .isNotNull()
+ .extracting(lifecycle -> lifecycle.getPreStop().getExec().getCommand())
+ .isEqualTo(
+ java.util.Arrays.asList(
+ "sh", "-c", "kill -USR2 1 2>/dev/null || true; sleep 40"));
+ }
+
+ @Test
+ void testSleepClampedToZeroForShortGracePeriods() {
+ flinkConfig.set(
+ KubernetesConfigOptions.TASK_MANAGER_TERMINATION_GRACE_PERIOD,
+ Duration.ofSeconds(2));
+
+ final FlinkPod resultFlinkPod = decorator.decorateFlinkPod(baseFlinkPod);
+
+ assertThat(
+ resultFlinkPod
+ .getMainContainer()
+ .getLifecycle()
+ .getPreStop()
+ .getExec()
+ .getCommand())
+ .isEqualTo(
+ java.util.Arrays.asList(
+ "sh", "-c", "kill -USR2 1 2>/dev/null || true; sleep 0"));
+ }
+
+ @Test
+ void testDoesNotOverwriteExistingGracePeriod() {
+ flinkConfig.set(
+ KubernetesConfigOptions.TASK_MANAGER_TERMINATION_GRACE_PERIOD, GRACE_PERIOD);
+ final FlinkPod podWithExistingGracePeriod =
+ new FlinkPod.Builder(baseFlinkPod)
+ .withPod(
+ new PodBuilder(baseFlinkPod.getPodWithoutMainContainer())
+ .editOrNewSpec()
+ .withTerminationGracePeriodSeconds(999L)
+ .endSpec()
+ .build())
+ .build();
+
+ final FlinkPod resultFlinkPod = decorator.decorateFlinkPod(podWithExistingGracePeriod);
+
+ assertThat(
+ resultFlinkPod
+ .getPodWithoutMainContainer()
+ .getSpec()
+ .getTerminationGracePeriodSeconds())
+ .isEqualTo(999L);
+ }
+
+ @Test
+ void testDoesNotOverwriteExistingPreStopHook() {
+ flinkConfig.set(
+ KubernetesConfigOptions.TASK_MANAGER_TERMINATION_GRACE_PERIOD, GRACE_PERIOD);
+ final FlinkPod podWithExistingPreStop =
+ new FlinkPod.Builder(baseFlinkPod)
+ .withMainContainer(
+ new ContainerBuilder(baseFlinkPod.getMainContainer())
+ .editOrNewLifecycle()
+ .withNewPreStop()
+ .withExec(
+ new ExecActionBuilder().withCommand("true").build())
+ .endPreStop()
+ .endLifecycle()
+ .build())
+ .build();
+
+ final FlinkPod resultFlinkPod = decorator.decorateFlinkPod(podWithExistingPreStop);
+
+ assertThat(
+ resultFlinkPod
+ .getMainContainer()
+ .getLifecycle()
+ .getPreStop()
+ .getExec()
+ .getCommand())
+ .containsExactly("true");
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 13af32c8884422..dbba8f5354a21a 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -1187,7 +1187,6 @@ protected Optional closeTaskManagerConnection(
ExceptionUtils.returnExceptionIfUnexpected(cause.getCause()));
ExceptionUtils.logExceptionIfExcepted(cause.getCause(), log);
- // TODO :: suggest failed task executor to stop itself
slotManager.unregisterTaskManager(workerRegistration.getInstanceID(), cause);
clusterPartitionTracker.processTaskExecutorShutdown(resourceID);
@@ -1199,7 +1198,14 @@ protected Optional closeTaskManagerConnection(
.getJobManagerGateway()
.disconnectTaskManager(resourceID, cause));
- workerRegistration.getTaskExecutorGateway().disconnectResourceManager(cause);
+ // Instruct the TaskExecutor to stop itself rather than reconnect: by the time we
+ // get here, the ResourceManager has already unregistered it and told every
+ // JobMaster to fail its tasks, so there is no cluster state left for this
+ // TaskExecutor to rejoin. This is a best-effort notification - if the TaskExecutor
+ // is unreachable (e.g. a genuine network partition), it cannot be delivered, and
+ // the TaskExecutor's own voluntary/graceful-termination path is the only other way
+ // it learns to stop.
+ workerRegistration.getTaskExecutorGateway().fenceAndStop(cause);
} else {
log.debug(
"No open TaskExecutor connection {}. Ignoring close TaskExecutor connection. Closing reason was: {}",
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index e5b9fc9b79d4ca..90ed49d900bd32 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -305,6 +305,13 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
@Nullable private UUID currentRegistrationTimeoutId;
+ /**
+ * Once set, this TaskExecutor no longer attempts to (re)connect to a ResourceManager, whether
+ * because it is voluntarily preparing for termination ({@link #prepareForTermination()}) or
+ * because a ResourceManager has already fenced it off ({@link #fenceAndStop}).
+ */
+ private volatile boolean terminationRequested = false;
+
private final Map>>
taskResultPartitionCleanupFuturesPerJob = CollectionUtil.newHashMapWithExpectedSize(8);
@@ -1431,6 +1438,60 @@ public void disconnectResourceManager(Exception cause) {
}
}
+ @Override
+ public void fenceAndStop(Exception cause) {
+ runAsync(() -> handleFenceAndStop(cause));
+ }
+
+ private void handleFenceAndStop(Exception cause) {
+ if (terminationRequested) {
+ return;
+ }
+ terminationRequested = true;
+ log.warn(
+ "ResourceManager instructed this TaskExecutor to stop ({}). Failing all locally "
+ + "running tasks now instead of waiting for each JobMaster to notice the "
+ + "loss of this TaskExecutor independently.",
+ cause.getMessage());
+ disconnectAndStopReconnectingToResourceManager(cause);
+ for (JobTable.Job job : jobTable.getJobs()) {
+ job.asConnection()
+ .ifPresent(
+ jobManagerConnection ->
+ disconnectJobManagerConnection(
+ jobManagerConnection, cause, true));
+ }
+ }
+
+ /**
+ * Best-effort, voluntary counterpart to {@link #fenceAndStop}: disconnects from the
+ * ResourceManager without waiting to be timed out, so the ResourceManager can free this
+ * TaskExecutor's slots immediately. Unlike {@link #fenceAndStop}, this does not fail currently
+ * running tasks - it is intended to be triggered ahead of a graceful pod termination, where the
+ * caller (see {@code TaskManagerRunner#prepareForTermination()}) separately bounds how long
+ * in-flight tasks are given to finish before the process exits.
+ */
+ public void prepareForTermination() {
+ runAsync(this::handlePrepareForTermination);
+ }
+
+ private void handlePrepareForTermination() {
+ if (terminationRequested) {
+ return;
+ }
+ terminationRequested = true;
+ log.info(
+ "Preparing for graceful termination: disconnecting from the ResourceManager "
+ + "instead of waiting to be timed out.");
+ disconnectAndStopReconnectingToResourceManager(
+ new FlinkException("TaskExecutor is preparing for graceful termination."));
+ }
+
+ private void disconnectAndStopReconnectingToResourceManager(Exception cause) {
+ closeResourceManagerConnection(cause);
+ resourceManagerAddress = null;
+ }
+
// ----------------------------------------------------------------------
// Other RPCs
// ----------------------------------------------------------------------
@@ -1535,6 +1596,9 @@ public CompletableFuture> requestProfilingList(Duratio
private void notifyOfNewResourceManagerLeader(
String newLeaderAddress, ResourceManagerId newResourceManagerId) {
+ if (terminationRequested) {
+ return;
+ }
resourceManagerAddress =
createResourceManagerAddress(newLeaderAddress, newResourceManagerId);
reconnectToResourceManager(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
index 29607c2c2f8130..0e58b3fd7e82d1 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
@@ -223,6 +223,21 @@ CompletableFuture heartbeatFromJobManager(
*/
void disconnectResourceManager(Exception cause);
+ /**
+ * Instructs this TaskExecutor to stop acting as a member of the cluster: it will no longer try
+ * to reconnect to the ResourceManager that issued this call, and it proactively fails the tasks
+ * it is currently running for every JobMaster it is connected to, rather than relying on each
+ * JobMaster to independently notice the loss of this TaskExecutor.
+ *
+ * The ResourceManager calls this once it has already decided, via heartbeat timeout or an
+ * explicit voluntary disconnect, that this TaskExecutor is no longer part of the cluster, so
+ * that the TaskExecutor stops producing further side effects for those tasks as soon as
+ * possible.
+ *
+ * @param cause the reason the ResourceManager is fencing off this TaskExecutor
+ */
+ void fenceAndStop(Exception cause);
+
/**
* Frees the slot with the given allocation ID.
*
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
index 5d33ddd65a05e9..c7351113156dfd 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGatewayDecoratorBase.java
@@ -186,6 +186,11 @@ public void disconnectResourceManager(Exception cause) {
originalGateway.disconnectResourceManager(cause);
}
+ @Override
+ public void fenceAndStop(Exception cause) {
+ originalGateway.fenceAndStop(cause);
+ }
+
@Override
public CompletableFuture freeSlot(
AllocationID allocationId, Throwable cause, Duration timeout) {
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
index c2970ff5e5e28a..5fc3b738a1fd3e 100644
--- 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
@@ -42,6 +42,11 @@ public CompletableFuture getTerminationFuture() {
return taskExecutor.getTerminationFuture();
}
+ @Override
+ public void prepareForTermination() {
+ taskExecutor.prepareForTermination();
+ }
+
@Override
public CompletableFuture closeAsync() {
return taskExecutor.closeAsync();
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 c8f0e24f73218c..31401b330fd0a0 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
@@ -75,6 +75,7 @@
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.ExecutorUtils;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.util.OperatingSystem;
import org.apache.flink.util.Reference;
import org.apache.flink.util.ShutdownHookUtil;
import org.apache.flink.util.StringUtils;
@@ -85,6 +86,7 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import sun.misc.Signal;
import javax.annotation.concurrent.GuardedBy;
@@ -306,6 +308,21 @@ public void start() throws Exception {
}
}
+ /**
+ * Best-effort request to prepare this TaskExecutor for an imminent, voluntary termination.
+ * Intended to be triggered by a Kubernetes {@code preStop} hook (via {@code SIGUSR2}, see
+ * {@link #registerGracefulTerminationSignalHandler}) ahead of the {@code SIGTERM} that ends the
+ * process, so the ResourceManager learns this TaskExecutor is leaving immediately instead of
+ * waiting for a heartbeat timeout.
+ */
+ public void prepareForTermination() {
+ synchronized (lock) {
+ if (taskExecutorService != null) {
+ taskExecutorService.prepareForTermination();
+ }
+ }
+ }
+
public void close() throws Exception {
try {
closeAsync().get();
@@ -499,6 +516,7 @@ public static int runTaskManager(Configuration configuration, PluginManager plug
configuration,
pluginManager,
TaskManagerRunner::createTaskExecutorService);
+ registerGracefulTerminationSignalHandler(taskManagerRunner);
taskManagerRunner.start();
} catch (Exception exception) {
throw new FlinkException("Failed to start the TaskManagerRunner.", exception);
@@ -562,6 +580,36 @@ public static void runTaskManagerProcessSecurely(Configuration configuration) {
// Static utilities
// --------------------------------------------------------------------------------------------
+ /**
+ * Registers a handler for {@code SIGUSR2} that triggers {@link #prepareForTermination()} on the
+ * given runner. This is the signal a Kubernetes {@code preStop} hook is expected to send (see
+ * the {@code kubernetes.taskmanager.termination-grace-period} option) ahead of the {@code
+ * SIGTERM} that Kubernetes sends once the {@code preStop} hook returns. Unlike {@link
+ * SignalHandler}, this handler does not delegate to a previous handler or terminate the process
+ * - {@code SIGUSR2} has no default JVM behavior, so this is purely additive.
+ *
+ * {@code SIGUSR2} is not available on Windows; the handler is simply not registered there,
+ * and graceful termination remains opt-in via a POSIX {@code preStop} hook regardless.
+ */
+ private static void registerGracefulTerminationSignalHandler(TaskManagerRunner runner) {
+ if (OperatingSystem.isWindows()) {
+ return;
+ }
+ try {
+ Signal.handle(
+ new Signal("USR2"),
+ signal -> {
+ LOG.info(
+ "RECEIVED SIGNAL {}: SIG{}. Preparing for graceful termination.",
+ signal.getNumber(),
+ signal.getName());
+ runner.prepareForTermination();
+ });
+ } catch (Exception e) {
+ LOG.info("Error while registering graceful termination signal handler", e);
+ }
+ }
+
public static TaskExecutorService createTaskExecutorService(
Configuration configuration,
ResourceID resourceID,
@@ -811,6 +859,15 @@ public interface TaskExecutorService extends AutoCloseableAsync {
void start();
CompletableFuture getTerminationFuture();
+
+ /**
+ * Best-effort request to prepare this TaskExecutor for an imminent, voluntary termination:
+ * disconnect from the ResourceManager without attempting to reconnect, instead of waiting
+ * to be timed out. Does not fail currently running tasks; callers that want a bounded drain
+ * window are expected to give tasks time to finish before the process is actually
+ * terminated.
+ */
+ void prepareForTermination();
}
public enum Result {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
index 11dc2e2c5321e4..b36d97bd20063c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
@@ -1933,6 +1933,88 @@ void testReconnectionAttemptIfExplicitlyDisconnected() throws Exception {
}
}
+ /**
+ * Tests that {@link TaskExecutor#prepareForTermination()} proactively disconnects from the
+ * ResourceManager, and - unlike an ordinary disconnect - does not attempt to reconnect
+ * afterwards.
+ */
+ @Test
+ void testPrepareForTerminationDisconnectsFromResourceManagerWithoutReconnecting()
+ throws Exception {
+ final TaskSlotTable taskSlotTable =
+ TaskSlotUtils.createTaskSlotTable(1, EXECUTOR_EXTENSION.getExecutor());
+ final UnresolvedTaskManagerLocation unresolvedTaskManagerLocation =
+ new LocalUnresolvedTaskManagerLocation();
+ final TaskExecutor taskExecutor =
+ createTaskExecutor(
+ new TaskManagerServicesBuilder()
+ .setTaskSlotTable(taskSlotTable)
+ .setUnresolvedTaskManagerLocation(unresolvedTaskManagerLocation)
+ .build());
+
+ taskExecutor.start();
+
+ try {
+ final TestingResourceManagerGateway testingResourceManagerGateway =
+ new TestingResourceManagerGateway();
+ final ClusterInformation clusterInformation = new ClusterInformation("foobar", 1234);
+ final CompletableFuture registrationResponseFuture =
+ CompletableFuture.completedFuture(
+ new TaskExecutorRegistrationSuccess(
+ new InstanceID(),
+ ResourceID.generate(),
+ clusterInformation,
+ null));
+ final BlockingQueue registrationQueue = new ArrayBlockingQueue<>(1);
+ final CompletableFuture disconnectTaskManagerFuture =
+ new CompletableFuture<>();
+ final OneShotLatch slotReportReceived = new OneShotLatch();
+
+ testingResourceManagerGateway.setRegisterTaskExecutorFunction(
+ taskExecutorRegistration -> {
+ registrationQueue.offer(taskExecutorRegistration.getResourceId());
+ return registrationResponseFuture;
+ });
+ testingResourceManagerGateway.setDisconnectTaskExecutorConsumer(
+ tuple -> disconnectTaskManagerFuture.complete(tuple.f0));
+ testingResourceManagerGateway.setSendSlotReportFunction(
+ ignored -> {
+ slotReportReceived.trigger();
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ });
+ rpc.registerGateway(
+ testingResourceManagerGateway.getAddress(), testingResourceManagerGateway);
+
+ resourceManagerLeaderRetriever.notifyListener(
+ testingResourceManagerGateway.getAddress(),
+ testingResourceManagerGateway.getFencingToken().toUUID());
+
+ final ResourceID firstRegistrationAttempt = registrationQueue.take();
+ assertThat(firstRegistrationAttempt)
+ .isEqualTo(unresolvedTaskManagerLocation.getResourceID());
+
+ // Wait until the TaskExecutor has actually finished establishing the ResourceManager
+ // connection (registration alone does not guarantee this has happened yet) - otherwise
+ // prepareForTermination() may run before there is a connection to close.
+ slotReportReceived.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
+
+ taskExecutor.prepareForTermination();
+
+ final ResourceID disconnectedResourceId =
+ disconnectTaskManagerFuture.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ assertThat(disconnectedResourceId)
+ .isEqualTo(unresolvedTaskManagerLocation.getResourceID());
+
+ assertThat(registrationQueue.poll(500L, TimeUnit.MILLISECONDS))
+ .withFailMessage(
+ "A TaskExecutor preparing for termination must not try to "
+ + "re-register with the ResourceManager it just disconnected from.")
+ .isNull();
+ } finally {
+ RpcUtils.terminateRpcEndpoint(taskExecutor);
+ }
+ }
+
/**
* Tests that the {@link TaskExecutor} sends the initial slot report after it registered at the
* ResourceManager.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
index cf1b79c68e643a..860f90a28806a7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
@@ -100,6 +100,8 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
private final Consumer disconnectResourceManagerConsumer;
+ private final Consumer fenceAndStopConsumer;
+
private final Function> cancelTaskFunction;
private final Supplier> canBeReleasedSupplier;
@@ -158,6 +160,7 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
Consumer freeInactiveSlotsConsumer,
Function> heartbeatResourceManagerFunction,
Consumer disconnectResourceManagerConsumer,
+ Consumer fenceAndStopConsumer,
Function> cancelTaskFunction,
Supplier> canBeReleasedSupplier,
BiConsumer> releasePartitionsConsumer,
@@ -195,6 +198,7 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
this.freeInactiveSlotsConsumer = Preconditions.checkNotNull(freeInactiveSlotsConsumer);
this.heartbeatResourceManagerFunction = heartbeatResourceManagerFunction;
this.disconnectResourceManagerConsumer = disconnectResourceManagerConsumer;
+ this.fenceAndStopConsumer = fenceAndStopConsumer;
this.cancelTaskFunction = cancelTaskFunction;
this.canBeReleasedSupplier = canBeReleasedSupplier;
this.releasePartitionsConsumer = releasePartitionsConsumer;
@@ -324,6 +328,11 @@ public void disconnectResourceManager(Exception cause) {
disconnectResourceManagerConsumer.accept(cause);
}
+ @Override
+ public void fenceAndStop(Exception cause) {
+ fenceAndStopConsumer.accept(cause);
+ }
+
@Override
public CompletableFuture freeSlot(
AllocationID allocationId, Throwable cause, Duration timeout) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
index 988dcdf1f3c0e1..f2466e6de8ff5b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
@@ -88,6 +88,7 @@ public class TestingTaskExecutorGatewayBuilder {
NOOP_HEARTBEAT_RESOURCE_MANAGER_FUNCTION = ignored -> FutureUtils.completedVoidFuture();
private static final Consumer NOOP_DISCONNECT_RESOURCE_MANAGER_CONSUMER =
ignored -> {};
+ private static final Consumer NOOP_FENCE_AND_STOP_CONSUMER = ignored -> {};
private static final Function>
NOOP_CANCEL_TASK_FUNCTION =
ignored -> CompletableFuture.completedFuture(Acknowledge.get());
@@ -156,6 +157,7 @@ public class TestingTaskExecutorGatewayBuilder {
NOOP_HEARTBEAT_RESOURCE_MANAGER_FUNCTION;
private Consumer disconnectResourceManagerConsumer =
NOOP_DISCONNECT_RESOURCE_MANAGER_CONSUMER;
+ private Consumer fenceAndStopConsumer = NOOP_FENCE_AND_STOP_CONSUMER;
private Function> cancelTaskFunction =
NOOP_CANCEL_TASK_FUNCTION;
private Supplier> canBeReleasedSupplier =
@@ -269,6 +271,12 @@ public TestingTaskExecutorGatewayBuilder setDisconnectResourceManagerConsumer(
return this;
}
+ public TestingTaskExecutorGatewayBuilder setFenceAndStopConsumer(
+ Consumer fenceAndStopConsumer) {
+ this.fenceAndStopConsumer = fenceAndStopConsumer;
+ return this;
+ }
+
public TestingTaskExecutorGatewayBuilder setCancelTaskFunction(
Function> cancelTaskFunction) {
this.cancelTaskFunction = cancelTaskFunction;
@@ -358,6 +366,7 @@ public TestingTaskExecutorGateway createTestingTaskExecutorGateway() {
freeInactiveSlotsConsumer,
heartbeatResourceManagerFunction,
disconnectResourceManagerConsumer,
+ fenceAndStopConsumer,
cancelTaskFunction,
canBeReleasedSupplier,
releasePartitionsConsumer,
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
index 53d9627f8e796e..6b522fcd9f4d00 100644
--- 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
@@ -23,14 +23,17 @@
/** Testing implementation of {@link TaskManagerRunner.TaskExecutorService}. */
public class TestingTaskExecutorService implements TaskManagerRunner.TaskExecutorService {
private final Runnable startRunnable;
+ private final Runnable prepareForTerminationRunnable;
private final CompletableFuture terminationFuture;
private final boolean completeTerminationFutureOnClose;
private TestingTaskExecutorService(
Runnable startRunnable,
+ Runnable prepareForTerminationRunnable,
CompletableFuture terminationFuture,
boolean completeTerminationFutureOnClose) {
this.startRunnable = startRunnable;
+ this.prepareForTerminationRunnable = prepareForTerminationRunnable;
this.terminationFuture = terminationFuture;
this.completeTerminationFutureOnClose = completeTerminationFutureOnClose;
}
@@ -45,6 +48,11 @@ public CompletableFuture getTerminationFuture() {
return terminationFuture;
}
+ @Override
+ public void prepareForTermination() {
+ prepareForTerminationRunnable.run();
+ }
+
@Override
public CompletableFuture closeAsync() {
if (completeTerminationFutureOnClose) {
@@ -60,6 +68,7 @@ public static Builder newBuilder() {
/** Builder for {@link TestingTaskExecutorService}. */
public static final class Builder {
private Runnable startRunnable = () -> {};
+ private Runnable prepareForTerminationRunnable = () -> {};
private CompletableFuture terminationFuture = new CompletableFuture<>();
private boolean completeTerminationFutureOnClose = true;
@@ -68,6 +77,11 @@ public Builder setStartRunnable(Runnable startRunnable) {
return this;
}
+ public Builder setPrepareForTerminationRunnable(Runnable prepareForTerminationRunnable) {
+ this.prepareForTerminationRunnable = prepareForTerminationRunnable;
+ return this;
+ }
+
public Builder setTerminationFuture(CompletableFuture terminationFuture) {
this.terminationFuture = terminationFuture;
return this;
@@ -80,7 +94,10 @@ public Builder withManualTerminationFutureCompletion() {
TestingTaskExecutorService build() {
return new TestingTaskExecutorService(
- startRunnable, terminationFuture, completeTerminationFutureOnClose);
+ startRunnable,
+ prepareForTerminationRunnable,
+ terminationFuture,
+ completeTerminationFutureOnClose);
}
}
}