[kubernetes] Graceful TaskManager termination + fix wrong-signal TODO in ResourceManager

[gangavhi]

What is the purpose of the change

On Kubernetes, a TaskManager pod is handled the same way whether it's lost voluntarily (node
drain, spot/preemptible reclaim, rolling upgrade) or crashes outright: the ResourceManager only
learns it's gone once the heartbeat times out (50s by default), and SIGTERM tears the process
down immediately with no chance for running tasks to make further progress.

This PR adds an opt-in kubernetes.taskmanager.termination-grace-period option that gives a
TaskManager pod a real, bounded grace window on voluntary eviction, and separately fixes a
long-standing TODO in ResourceManager where the "notify the TaskExecutor to stop" step existed
but sent the wrong instruction.

Not a Jira ticket yet — happy to file one (or have this PR converted into the trigger for one)
based on maintainer feedback on scope/approach; wanted to put the code up for review first rather
than guess at a title before there's been any discussion.

Brief change log

• kubernetes.taskmanager.termination-grace-period (new ConfigOption<Duration>) +
  TerminationGracePeriodDecorator: sets terminationGracePeriodSeconds and a preStop hook on
  the TaskManager pod, never overriding a value already present in the user's pod template.
• The preStop hook sends SIGUSR2 to the main process, which TaskManagerRunner routes to the
  new TaskExecutor#prepareForTermination(): disconnects from the ResourceManager proactively
  instead of waiting to be timed out, so the slot is freed immediately. The hook then sleeps for
  most of the remaining grace period before returning (Kubernetes only sends SIGTERM once
  preStop returns), so already-running tasks keep making progress in the meantime.
• ResourceManager#closeTaskManagerConnection carried
  // TODO :: suggest failed task executor to stop itself. It already calls back into the
  presumed-dead TaskExecutor via TaskExecutorGateway#disconnectResourceManager — but that RPC
  tells the TaskExecutor to reconnect, which is the wrong instruction for a TaskExecutor the
  ResourceManager has already unregistered and told every JobMaster to fail. Adds
  TaskExecutorGateway#fenceAndStop, called here instead: tells the TaskExecutor to stop trying
  to reconnect and to proactively fail its own currently-running tasks.
• Docs: new "Graceful TaskManager Termination" section in native_kubernetes.md; regenerated
  kubernetes_config_configuration.html.

Deliberately out of scope, to keep this reviewable: equivalent JobManager-side graceful
shutdown (needs its own SIGUSR2 handler first — sending it unhandled terminates a process by
POSIX default, the opposite of graceful), and a TaskManager-side proactive self-fencing/isolation
timeout for the case where the ResourceManager's own notification can't be delivered at all (a
true network partition) — that changes failure-detection timing, not just teardown messaging,
and seems like it deserves its own dev@ discussion rather than being bundled here.

Verifying this change

• New unit tests: TerminationGracePeriodDecoratorTest (pod-spec assertions: option unset is a
  no-op, grace period + preStop set correctly, neither overrides an existing pod-template value,
  sleep duration clamps to zero for very short grace periods), plus two new tests in
  TaskExecutorTest covering prepareForTermination() (disconnects and does not reconnect) and
  fenceAndStop (fails locally-running tasks and disconnects the JobMaster connection, reusing
  the existing runJobManagerHeartbeatTest harness).
• Full flink-kubernetes module test suite and the full TaskExecutorTest /
  TaskManagerRunnerTest classes pass (checked for regressions since a shared gateway interface,
  TaskExecutorGateway, gained a new method).
• ./mvnw spotless:check and ./mvnw checkstyle:check clean on touched modules.
• Manually verified end-to-end on a local kind cluster with a build of this branch: submitted a
  checkpointed streaming job, then kubectl delete pod on the TaskManager. Confirmed via
  timestamped logs that the ResourceManager's proactive-disconnect log line lands within ~1s of
  the SIGUSR2 signal (well before the 50s heartbeat timeout would fire), and that the pod (and
  its task) stays alive for the configured grace window before SIGTERM arrives — compared
  against a baseline session cluster without the option set, where SIGTERM fires the instant
  kubectl delete pod is issued today.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): (yes) — adds
  one new ConfigOption to KubernetesConfigOptions (@PublicEvolving); purely additive,
  defaults to today's behavior when unset.
• The serializers: (no)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes) — TaskManager pod teardown on Kubernetes; see change log above.
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (yes)
• If yes, how is the feature documented? (docs, generated config reference)

---

Was generative AI tooling used to co-author this PR?

• Yes (Claude Code / Claude Sonnet 5)

I worked with Claude Code through the design, implementation, and verification of this change,
including tracing the existing heartbeat/RM/TM code paths to scope it, and reviewing the kind
cluster test results (which surfaced and required fixing two real bugs during development: a
preStop script that fired the signal but didn't actually block for the grace period, and a race
condition in one of the new unit tests). I'm responsible for the correctness of everything in this
PR regardless of tooling used.

[flinkbot]

CI report:

• b190e69 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[spuru9]

Can you file a JIRA for this change and work on fixing the CI failures before opening the PR for review.