[FLINK-40019][core][runtime] Support per-job delegation tokens

docs/layouts/shortcodes/generated/security_configuration.html

@@ -44,6 +44,12 @@
             <td>Double</td>
             <td>Ratio of the tokens's expiration time when new credentials should be re-obtained.</td>
         </tr>
+        <tr>
+            <td><h5>security.delegation.tokens.reobtain.cooldown</h5></td>
+            <td style="word-wrap: break-word;">30 s</td>
+            <td>Duration</td>
+            <td>Minimum time between two consecutive on-demand token re-obtain cycles, such as those triggered when a job is registered. Requests arriving within the cooldown are coalesced and deferred until it elapses. Does not affect the periodic renewal.</td>
+        </tr>
         <tr>
             <td><h5>security.kerberos.access.hadoopFileSystems</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

docs/layouts/shortcodes/generated/security_delegation_token_section.html

@@ -32,6 +32,12 @@
             <td>Double</td>
             <td>Ratio of the tokens's expiration time when new credentials should be re-obtained.</td>
         </tr>
+        <tr>
+            <td><h5>security.delegation.tokens.reobtain.cooldown</h5></td>
+            <td style="word-wrap: break-word;">30 s</td>
+            <td>Duration</td>
+            <td>Minimum time between two consecutive on-demand token re-obtain cycles, such as those triggered when a job is registered. Requests arriving within the cooldown are coalesced and deferred until it elapses. Does not affect the periodic renewal.</td>
+        </tr>
         <tr>
             <td><h5>security.delegation.token.provider.&lt;serviceName&gt;.enabled</h5></td>
             <td style="word-wrap: break-word;">true</td>

flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java

@@ -180,8 +180,19 @@ public class SecurityOptions {
                     .withDescription(
                             "Ratio of the tokens's expiration time when new credentials should be re-obtained.");
 
-    @Documentation.SuffixOption(DELEGATION_TOKEN_PROVIDER_PREFIX)
     @Documentation.Section(value = Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 5)
+    public static final ConfigOption<Duration> DELEGATION_TOKENS_REOBTAIN_COOLDOWN =
+            key("security.delegation.tokens.reobtain.cooldown")
+                    .durationType()
+                    .defaultValue(Duration.ofSeconds(30))
+                    .withDescription(
+                            "Minimum time between two consecutive on-demand token re-obtain "
+                                    + "cycles, such as those triggered when a job is registered. "
+                                    + "Requests arriving within the cooldown are coalesced and "
+                                    + "deferred until it elapses. Does not affect the periodic renewal.");
+
+    @Documentation.SuffixOption(DELEGATION_TOKEN_PROVIDER_PREFIX)
+    @Documentation.Section(value = Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 6)
     public static final ConfigOption<Boolean> DELEGATION_TOKEN_PROVIDER_ENABLED =
             key("enabled")
                     .booleanType()

flink-core/src/main/java/org/apache/flink/core/security/token/DelegationTokenManagerCallback.java

@@ -0,0 +1,42 @@
+/*
+ * 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.core.security.token;
+
+import org.apache.flink.annotation.Experimental;
+
+/**
+ * Handed to a {@link DelegationTokenProvider} at {@link
+ * DelegationTokenProvider#init(org.apache.flink.configuration.Configuration,
+ * DelegationTokenManagerCallback) init} time, giving the provider a way to ask the delegation token
+ * manager to re-obtain tokens. The provider may retain the callback and invoke it later, outside
+ * the {@code init}/{@code registerJob} call stack.
+ */
+@Experimental
+public interface DelegationTokenManagerCallback {
+
+    /**
+     * Requests an asynchronous token re-obtain and redistribution to all receivers, bringing the
+     * next obtain cycle forward instead of waiting for the periodic renewal.
+     *
+     * <p>May be called from any thread at any time after {@code init}. The manager coalesces
+     * requests and may apply a cooldown, so a call does not necessarily map to one obtain cycle.
+     * Returns immediately and does not wait for completion.
+     */
+    void reobtainDelegationTokens();
+}

flink-core/src/main/java/org/apache/flink/core/security/token/DelegationTokenProvider.java

@@ -19,6 +19,7 @@
 package org.apache.flink.core.security.token;
 
 import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.JobID;
 import org.apache.flink.configuration.Configuration;
 
 import java.util.Optional;
@@ -28,6 +29,19 @@
  * DelegationTokenManager through service loader. Basically the implementation of this interface is
  * responsible to produce the serialized form of tokens which will be handled by {@link
  * DelegationTokenReceiver} instances both on JobManager and TaskManager side.
+ *
+ * <p><b>Threading contract.</b> A single instance per provider implementation is created and {@link
+ * #init(Configuration, DelegationTokenManagerCallback) initialized} once and then shared for the
+ * lifetime of the manager. {@link #obtainDelegationTokens()} usually runs on the manager's IO
+ * executor, but the first cycle runs on the thread that starts the manager (the ResourceManager
+ * main thread) and one-shot obtains run on the caller's thread, so implementations must not assume
+ * a particular thread. {@link #registerJob(JobID, Configuration)} and {@link #unregisterJob(JobID)}
+ * are invoked from the ResourceManager main thread. These can therefore run concurrently with
+ * {@link #obtainDelegationTokens()}. {@link
+ * DelegationTokenManagerCallback#reobtainDelegationTokens()} may be invoked from any thread.
+ * Implementations must keep any per-job state thread-safe, and {@code registerJob}/{@code
+ * unregisterJob} must be non-blocking so they do not stall the ResourceManager — defer real work to
+ * {@link #obtainDelegationTokens()}.
  */
 @Experimental
 public interface DelegationTokenProvider {
@@ -75,6 +89,25 @@ default String serviceConfigPrefix() {
      */
     void init(Configuration configuration) throws Exception;
 
+    /**
+     * Called by DelegationTokenManager to initialize the provider after construction, additionally
+     * handing it a {@link DelegationTokenManagerCallback} it can use to request a token re-obtain.
+     *
+     * <p>This is the entry point the manager actually calls. The default implementation ignores the
+     * callback and delegates to {@link #init(Configuration)}, so providers that do not need to
+     * trigger re-obtains keep implementing only {@link #init(Configuration)}. A provider that wants
+     * to request re-obtains overrides this method, retains the callback, and invokes {@link
+     * DelegationTokenManagerCallback#reobtainDelegationTokens()} when needed.
+     *
+     * @param configuration Configuration to initialize the provider.
+     * @param callback Used to ask the manager to re-obtain tokens. May be retained and called
+     *     later.
+     */
+    default void init(Configuration configuration, DelegationTokenManagerCallback callback)
+            throws Exception {
+        init(configuration);
+    }
+
     /**
      * Return whether delegation tokens are required for this service.
      *
@@ -88,4 +121,69 @@ default String serviceConfigPrefix() {
      * @return the obtained delegation tokens.
      */
     ObtainedDelegationTokens obtainDelegationTokens() throws Exception;
+
+    /**
+     * Called with the job's configuration when its JobMaster registers with the ResourceManager.
+     * Re-registration may occur while the job's tasks are running.
+     *
+     * <p>This notification does not gate job initialization: operator coordinators may already have
+     * started. Successful registration does not imply that tokens have been obtained or
+     * distributed.
+     *
+     * <p>To get the job's tokens distributed without waiting for the periodic renewal, call {@link
+     * DelegationTokenManagerCallback#reobtainDelegationTokens()} on the callback handed to {@link
+     * #init(Configuration, DelegationTokenManagerCallback)} to request an asynchronous obtain
+     * cycle, subject to the configured cooldown.
+     *
+     * <p>A provider that requests a re-obtain must record this job's per-job state <em>before</em>
+     * invoking {@link DelegationTokenManagerCallback#reobtainDelegationTokens()}. That call merely
+     * schedules (or coalesces into) an obtain cycle that runs later on another thread. Recording
+     * first establishes the happens-before that lets the serving cycle observe this job's state.
+     * Recording afterwards races with the cycle and the job's tokens may be skipped until the next
+     * periodic renewal.
+     *
+     * <p>Must be idempotent: it may be called more than once for the same {@code jobId} (e.g. on
+     * JobManager or ResourceManager failover, when the JobMaster re-registers).
+     *
+     * <p>Should not throw: an unchecked exception or linkage error rejects the current registration
+     * attempt. If the manager does not currently track a successful registration for this job, it
+     * calls {@link #unregisterJob(JobID)} on all providers to attempt rollback. Otherwise, it keeps
+     * the existing registration and does not invoke {@code unregisterJob} for that failure, because
+     * the job's tasks may still be running. Keeping the registration does not undo changes
+     * providers made during the failed attempt.
+     *
+     * <p>Prefer deferring token retrieval to the retrying {@link #obtainDelegationTokens()} cycle
+     * so transient fetch failures do not prevent registration.
+     *
+     * @param jobId The job id of the job.
+     * @param jobConfiguration The job configuration.
+     */
+    default void registerJob(JobID jobId, Configuration jobConfiguration) {}
+
+    /**
+     * Called to release per-job state when a job is removed, a registration attempt is rolled back,
+     * or the manager stops its current session. A job is removed when it reaches a globally
+     * terminal state or its job-leader registration times out. Must be idempotent and should not
+     * throw.
+     *
+     * <p>Exceptions and linkage errors are caught and logged, so cleanup continues for the other
+     * providers. The manager removes the job from its tracking even if cleanup fails and does not
+     * retain it for a later cleanup attempt. Providers are responsible for releasing any remaining
+     * state in {@link #close()}.
+     *
+     * @param jobId The job id of the job.
+     */
+    default void unregisterJob(JobID jobId) {}
+
+    /**
+     * Closes the provider. Any resources should be released.
+     *
+     * <p>Called at most once, when the manager is closed at process shutdown. It is not called on
+     * ResourceManager leadership changes, those only stop and restart the manager's obtain session
+     * and the provider instance stays in use. An obtain cycle started just before shutdown may
+     * still be running, so {@code close()} may overlap an in-flight {@link
+     * #obtainDelegationTokens()} and implementations must release resources in a way that is safe
+     * with respect to that overlap.
+     */
+    default void close() {}
 }

flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java

@@ -498,6 +498,16 @@ protected CompletableFuture<Void> stopClusterServices(boolean cleanupHaData) {
 
             final Collection<CompletableFuture<Void>> terminationFutures = new ArrayList<>(3);
 
+            if (delegationTokenManager != null) {
+                try {
+                    // Terminal teardown of the delegation token providers. The per-session
+                    // manager stop() already ran when the ResourceManager component closed.
+                    delegationTokenManager.close();
+                } catch (Throwable t) {
+                    exception = ExceptionUtils.firstOrSuppressed(t, exception);
+                }
+            }
+
             if (blobServer != null) {
                 try {
                     blobServer.close();

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java

@@ -1599,6 +1599,7 @@ protected CompletableFuture<RegistrationResponse> invokeRegistration(
                             jobManagerResourceID,
                             jobManagerRpcAddress,
                             jobID,
+                            executionPlan.getJobConfiguration(),
                             timeout);
                 }
             };

flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java

@@ -1372,6 +1372,17 @@ private void terminateMiniClusterServices(boolean cleanupHaData) throws Exceptio
         Exception exception = null;
 
         synchronized (lock) {
+            if (delegationTokenManager != null) {
+                try {
+                    // Terminal teardown of the delegation token providers. The per-session
+                    // manager stop() already ran when the ResourceManager component closed.
+                    delegationTokenManager.close();
+                } catch (Exception e) {
+                    exception = ExceptionUtils.firstOrSuppressed(e, exception);
+                }
+                delegationTokenManager = null;
+            }
+
             if (blobCacheService != null) {
                 try {
                     blobCacheService.close();

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java

@@ -22,6 +22,7 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ThreadDumpMode;
 import org.apache.flink.runtime.blob.TransientBlobKey;
 import org.apache.flink.runtime.blocklist.BlockedNode;
@@ -116,8 +117,8 @@
  * <p>It offers the following methods as part of its rpc interface to interact with him remotely:
  *
  * <ul>
- *   <li>{@link #registerJobMaster(JobMasterId, ResourceID, String, JobID, Duration)} registers a
- *       {@link JobMaster} at the resource manager
+ *   <li>{@link #registerJobMaster(JobMasterId, ResourceID, String, JobID, Configuration, Duration)}
+ *       registers a {@link JobMaster} at the resource manager
  * </ul>
  */
 public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
@@ -367,12 +368,14 @@ public CompletableFuture<RegistrationResponse> registerJobMaster(
             final ResourceID jobManagerResourceId,
             final String jobManagerAddress,
             final JobID jobId,
+            final Configuration jobConfiguration,
             final Duration timeout) {
 
         checkNotNull(jobMasterId);
         checkNotNull(jobManagerResourceId);
         checkNotNull(jobManagerAddress);
         checkNotNull(jobId);
+        checkNotNull(jobConfiguration);
 
         try (MdcCloseable ignored = MdcUtils.withContext(MdcUtils.asContextData(jobId))) {
             if (!jobLeaderIdService.containsJob(jobId)) {
@@ -428,6 +431,20 @@ public CompletableFuture<RegistrationResponse> registerJobMaster(
                             jobMasterIdFuture,
                             (JobMasterGateway jobMasterGateway, JobMasterId leadingJobMasterId) -> {
                                 if (Objects.equals(leadingJobMasterId, jobMasterId)) {
+                                    // Reject a failed provider registration before installing the
+                                    // JobMaster registration. Report plugin linkage errors as
+                                    // registration failures too.
+                                    try {
+                                        delegationTokenManager.registerJob(jobId, jobConfiguration);
+                                    } catch (Exception | LinkageError e) {
+                                        return new RegistrationResponse.Failure(
+                                                new FlinkException(
+                                                        "Failed to register job "
+                                                                + jobId
+                                                                + " with the delegation token "
+                                                                + "manager",
+                                                        e));
+                                    }
                                     return registerJobMasterInternal(
                                             jobMasterGateway,
                                             jobId,
@@ -1224,6 +1241,15 @@ protected void removeJob(JobID jobId, Exception cause) {
         if (jobManagerRegistrations.containsKey(jobId)) {
             closeJobManagerConnection(jobId, ResourceRequirementHandling.CLEAR, cause);
         }
+
+        try {
+            delegationTokenManager.unregisterJob(jobId);
+        } catch (Exception e) {
+            log.warn(
+                    "Could not properly remove the job {} from the delegation token manager.",
+                    jobId,
+                    e);
+        }
     }
 
     protected void jobLeaderLostLeadership(JobID jobId, JobMasterId oldJobMasterId) {

flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java

@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ThreadDumpMode;
 import org.apache.flink.runtime.blob.BlobServer;
 import org.apache.flink.runtime.blob.TransientBlobKey;
@@ -62,12 +63,15 @@ public interface ResourceManagerGateway
         extends FencedRpcGateway<ResourceManagerId>, ClusterPartitionManager, BlocklistListener {
 
     /**
-     * Register a {@link JobMaster} at the resource manager.
+     * Register a {@link JobMaster} at the resource manager, supplying the job's {@link
+     * Configuration} so implementations can perform per-job initialization (e.g. obtaining
+     * job-scoped delegation tokens).
      *
      * @param jobMasterId The fencing token for the JobMaster leader
      * @param jobMasterResourceId The resource ID of the JobMaster that registers
      * @param jobMasterAddress The address of the JobMaster that registers
      * @param jobId The Job ID of the JobMaster that registers
+     * @param jobConfiguration The job's configuration, used for per-job initialization
      * @param timeout Timeout for the future to complete
      * @return Future registration response
      */
@@ -76,6 +80,7 @@ CompletableFuture<RegistrationResponse> registerJobMaster(
             ResourceID jobMasterResourceId,
             String jobMasterAddress,
             JobID jobId,
+            Configuration jobConfiguration,
             @RpcTimeout Duration timeout);
 
     /**