From b190e69a89542df1843c103613f09d9f98aa68ae Mon Sep 17 00:00:00 2001 From: gangavhi Date: Sun, 12 Jul 2026 15:47:01 -0500 Subject: [PATCH] [kubernetes] Graceful TaskManager termination + fix wrong-signal TODO in ResourceManager Adds kubernetes.taskmanager.termination-grace-period: when set, a new TerminationGracePeriodDecorator sets terminationGracePeriodSeconds and a preStop hook (SIGUSR2) on the TaskManager pod, never overriding a user-supplied pod template. SIGUSR2 triggers TaskExecutor#prepareForTermination(), which proactively disconnects from the ResourceManager instead of waiting to be heartbeat-timed-out, so the slot is freed immediately. The preStop hook then sleeps for most of the remaining grace period before returning, so already-running tasks keep making progress instead of being killed the instant the pod is scheduled for deletion (SIGTERM only follows once preStop returns). Separately, ResourceManager#closeTaskManagerConnection carried a "// TODO :: suggest failed task executor to stop itself" comment. Tracing the actual behavior showed the ResourceManager already calls back into the presumed-dead TaskExecutor here (TaskExecutorGateway#disconnectResourceManager) but that RPC tells it to reconnect, which is the wrong instruction for a TaskExecutor the ResourceManager has already unregistered and told every JobMaster to fail. Adds TaskExecutorGateway#fenceAndStop, which the ResourceManager now calls instead: it tells the TaskExecutor to stop reconnecting and to proactively fail its own currently-running tasks. Both changes are additive: the pod-spec change is opt-in via the new config option, and fenceAndStop only changes the RPC sent on a teardown path the ResourceManager has already committed to - no change to failure-detection timing or semantics. JobManager-side graceful shutdown is deliberately out of scope here: it would need its own SIGUSR2 handler first, since sending that signal unhandled terminates a process by POSIX default. Verified with unit tests and a live kind cluster test: kubectl delete pod on a TaskManager now keeps the pod's task alive for the configured grace window before SIGTERM arrives (previously ~0s with no preStop hook), and the ResourceManager's proactive-disconnect log line arrives within ~1s of the signal. Co-Authored-By: Claude Sonnet 5 --- .../resource-providers/native_kubernetes.md | 26 +++ .../kubernetes_config_configuration.html | 6 + .../KubernetesConfigOptions.java | 21 +++ .../TerminationGracePeriodDecorator.java | 143 ++++++++++++++++ .../factory/KubernetesTaskManagerFactory.java | 7 +- .../TerminationGracePeriodDecoratorTest.java | 152 ++++++++++++++++++ .../resourcemanager/ResourceManager.java | 10 +- .../runtime/taskexecutor/TaskExecutor.java | 64 ++++++++ .../taskexecutor/TaskExecutorGateway.java | 15 ++ .../TaskExecutorGatewayDecoratorBase.java | 5 + .../TaskExecutorToServiceAdapter.java | 5 + .../taskexecutor/TaskManagerRunner.java | 57 +++++++ .../taskexecutor/TaskExecutorTest.java | 82 ++++++++++ .../TestingTaskExecutorGateway.java | 9 ++ .../TestingTaskExecutorGatewayBuilder.java | 9 ++ .../TestingTaskExecutorService.java | 19 ++- 16 files changed, 626 insertions(+), 4 deletions(-) create mode 100644 flink-kubernetes/src/main/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecorator.java create mode 100644 flink-kubernetes/src/test/java/org/apache/flink/kubernetes/kubeclient/decorators/TerminationGracePeriodDecoratorTest.java diff --git a/docs/content/docs/deployment/resource-providers/native_kubernetes.md b/docs/content/docs/deployment/resource-providers/native_kubernetes.md index 2e0dd775c4007..8972e19d5d953 100644 --- a/docs/content/docs/deployment/resource-providers/native_kubernetes.md +++ b/docs/content/docs/deployment/resource-providers/native_kubernetes.md @@ -468,6 +468,32 @@ Configure the value of }}#kubernetes- It will help to achieve faster recovery. Notice that high availability should be enabled when starting standby JobManagers. +### Graceful TaskManager Termination + +By default, Flink's Kubernetes deployment reacts to a lost TaskManager the same way whether the +pod was evicted on purpose (a node drain, a spot/preemptible reclaim, a rolling upgrade) or crashed +outright: the ResourceManager only notices once the TaskManager's heartbeat times out (50 seconds +by default), and the TaskManager pod itself is torn down as soon as Kubernetes sends it `SIGTERM`, +with no chance for currently-running tasks to keep making progress in the meantime. + +Setting }}#kubernetes-taskmanager-termination-grace-period">kubernetes.taskmanager.termination-grace-period +changes this for voluntary evictions. It sets the pod's `terminationGracePeriodSeconds` and adds a +`preStop` lifecycle hook that: + +1. Signals the TaskExecutor to disconnect from the ResourceManager immediately, instead of waiting + to be timed out, so the ResourceManager can free its slots and schedule a replacement right away. +2. Keeps the pod alive for most of the configured grace period afterwards (a few seconds are + reserved at the end for the JVM's own shutdown sequence to run once Kubernetes sends + `SIGTERM`), so tasks that are already running continue to make progress - and get a chance to + complete an in-flight checkpoint - instead of being killed the instant the pod is scheduled for + deletion. + +This only takes effect on a graceful `kubectl delete pod` / node drain, where Kubernetes actually +runs `preStop` and waits out the grace period; it does not change how quickly Flink notices a pod +that has actually crashed or become unreachable, since there is no signal to send to a process +that is no longer running. It has no effect if the pod template already sets +`terminationGracePeriodSeconds` or a `preStop` hook. + ### Manual Resource Cleanup Flink uses [Kubernetes OwnerReference's](https://kubernetes.io/docs/concepts/workloads/controllers/garbage-collection/) to clean up all cluster components. diff --git a/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html b/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html index a3084585bf4cb..62be97b9179d1 100644 --- a/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html +++ b/docs/layouts/shortcodes/generated/kubernetes_config_configuration.html @@ -326,6 +326,12 @@ String Service account that is used by taskmanager within kubernetes cluster. The task manager uses this service account when watching config maps on the API server to retrieve leader address of jobmanager and resourcemanager. If not explicitly configured, config option 'kubernetes.service-account' will be used. + +
kubernetes.taskmanager.termination-grace-period
+ (none) + Duration + If set, the TaskManager pod's terminationGracePeriodSeconds is set to this value, and a preStop lifecycle hook is added that signals the TaskExecutor to proactively disconnect from the ResourceManager - instead of waiting to be timed out - so its slots are freed immediately rather than after the heartbeat timeout elapses. Left unset, no lifecycle hook is added and Kubernetes' own default grace period behavior applies. Has no effect if the pod template already configures terminationGracePeriodSeconds or a preStop hook. +
kubernetes.taskmanager.tolerations
(none) diff --git a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java index bac5cced79901..42398bf4bbc15 100644 --- a/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java +++ b/flink-kubernetes/src/main/java/org/apache/flink/kubernetes/configuration/KubernetesConfigOptions.java @@ -656,6 +656,27 @@ public class KubernetesConfigOptions { * #KUBERNETES_PERSISTENT_VOLUME_CLAIMS}. If you need different access modes for different PVCs, * consider using pod templates instead. */ + public static final ConfigOption TASK_MANAGER_TERMINATION_GRACE_PERIOD = + key("kubernetes.taskmanager.termination-grace-period") + .durationType() + .noDefaultValue() + .withDescription( + Description.builder() + .text( + "If set, the TaskManager pod's %s is set to this value, and a %s " + + "lifecycle hook is added that signals the TaskExecutor to " + + "proactively disconnect from the ResourceManager - instead of " + + "waiting to be timed out - so its slots are freed immediately " + + "rather than after the heartbeat timeout elapses. Left unset, " + + "no lifecycle hook is added and Kubernetes' own default grace " + + "period behavior applies. Has no effect if the pod template " + + "already configures %s or a %s hook.", + code("terminationGracePeriodSeconds"), + code("preStop"), + code("terminationGracePeriodSeconds"), + code("preStop")) + .build()); + public static final ConfigOption 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 0000000000000..68562229b77f7 --- /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 05ed8c91cbf4d..3896c7b9683ca 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 0000000000000..9c9e6dd4a3ab3 --- /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 13af32c888442..dbba8f5354a21 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 e5b9fc9b79d4c..90ed49d900bd3 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 29607c2c2f813..0e58b3fd7e82d 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 5d33ddd65a05e..c7351113156df 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 c2970ff5e5e28..5fc3b738a1fd3 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 c8f0e24f73218..31401b330fd0a 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 11dc2e2c5321e..b36d97bd20063 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 cf1b79c68e643..860f90a28806a 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 988dcdf1f3c0e..f2466e6de8ff5 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 53d9627f8e796..6b522fcd9f4d0 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); } } }