Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Savonitar wants to merge 6 commits into
base: master
Choose a base branch
from
+1,334 −56
Draft

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
84abc1d
[FLINK-40019][core][runtime] Support per-job delegation tokens
Savonitar Jun 27, 2026
90a9f03
[FLINK-40019][core][runtime] Polish comments and apply spotless forma…
Savonitar Jul 7, 2026
55b0566
[FLINK-40019][runtime] Harden the delegation token manager lifecycle
Savonitar Jul 7, 2026
e2dba9e
[FLINK-40019][core][runtime] Treat provider LinkageError like a failu…
Savonitar Jul 7, 2026
dfb3cd6
[FLINK-40019][runtime] Add job and provider in registration failure logs
Savonitar Jul 7, 2026
6716576
[FLINK-40019][runtime] Fix concurrency test and clamp negative renewa…
Savonitar Jul 7, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions docs/layouts/shortcodes/generated/security_configuration.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
6 changes: 6 additions & 0 deletions docs/layouts/shortcodes/generated/security_delegation_token_section.html
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down
13 changes: 12 additions & 1 deletion flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
42 changes: 42 additions & 0 deletions ...re/src/main/java/org/apache/flink/core/security/token/DelegationTokenManagerCallback.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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();
}
78 changes: 78 additions & 0 deletions flink-core/src/main/java/org/apache/flink/core/security/token/DelegationTokenProvider.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -28,6 +29,16 @@
* 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()} runs on the manager's IO executor,
* while {@link #registerJob(JobID, Configuration)} and {@link #unregisterJob(JobID)} are invoked
* from the ResourceManager main thread. These can therefore run concurrently. {@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 {
Expand Down Expand Up @@ -75,6 +86,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.
*
Expand All @@ -88,4 +118,52 @@ default String serviceConfigPrefix() {
* @return the obtained delegation tokens.
*/
ObtainedDelegationTokens obtainDelegationTokens() throws Exception;

/**
* Called when a job has started, before its tasks are scheduled, with its configuration.
*
* <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 immediate obtain cycle.
*
* <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: a thrown (unchecked) exception or linkage error rejects the job's
* registration (the job does not start) and triggers {@link #unregisterJob(JobID)} on all
* providers to roll back. Prefer deferring the real fetch to the (retrying) obtain cycle over a
* synchronous fetch, so a transient failure does not fail the job.
*
* @param jobId The job id of the job.
* @param jobConfiguration The job configuration.
*/
default void registerJob(JobID jobId, Configuration jobConfiguration) {}

/**
* Called when the job is being removed — it reached a globally terminal state, or its
* job-leader registration timed out — and its per-job state should be released. Must be
* idempotent. Exceptions and linkage errors are caught and logged by the framework (one
* provider's failure does not abort cleanup of the others), but implementations should still
* avoid throwing.
*
* @param jobId The job id of the job.
*/
default void unregisterJob(JobID jobId) {}

/**
* Stops the provider. Any resources should be closed.
*
* <p>Called once during manager shutdown. Note that an obtain-and-broadcast cycle started just
* before shutdown may still be running on another thread when this is invoked, so {@code
* stop()} may overlap an in-flight {@link #obtainDelegationTokens()}; implementations must
* release resources in a way that is safe with respect to that overlap.
*/
default void stop() {}
}
1 change: 1 addition & 0 deletions flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1599,6 +1599,7 @@ protected CompletableFuture<RegistrationResponse> invokeRegistration(
jobManagerResourceID,
jobManagerRpcAddress,
jobID,
executionPlan.getJobConfiguration(),
timeout);
}
};
Expand Down
27 changes: 25 additions & 2 deletions flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.blocklist.BlockedNode;
import org.apache.flink.runtime.blocklist.BlocklistContext;
Expand Down Expand Up @@ -115,8 +116,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>
Expand Down Expand Up @@ -366,12 +367,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)) {
Expand Down Expand Up @@ -427,6 +430,17 @@ public CompletableFuture<RegistrationResponse> registerJobMaster(
jobMasterIdFuture,
(JobMasterGateway jobMasterGateway, JobMasterId leadingJobMasterId) -> {
if (Objects.equals(leadingJobMasterId, jobMasterId)) {
// Register with the delegation token manager first. A
// provider failure rejects this registration so the job
// never starts without the tokens it requires. LinkageError
// is included so a provider plugin classpath failure becomes
// a proper registration failure instead of an exceptionally
// completed future whose cause is only visible at debug level.
try {
delegationTokenManager.registerJob(jobId, jobConfiguration);
} catch (Exception | LinkageError e) {
return new RegistrationResponse.Failure(e);

@gaborgsomogyi gaborgsomogyi Jul 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we see this exception somehwere? I'm pretty sure if this happens oncall guys want to know what happened.

@Savonitar Savonitar Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

oncall guys

I'm something of an on-call guy myself...

Yes, it is logged in https://github.com/apache/flink/pull/28639/changes#diff-d32f89982ba30269ab9438d993d9deb3b29d046792b6cc5ea35a277eaacfb287R629 (flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java registerJob()

LOG.error("Failed to register job {}", jobId, e);

Example:

ERROR org.apache.flink.runtime.security.token.DefaultDelegationTokenManager - Failed to register job 50bfee4c5d12d0154333bc23410c92c3q9
  java.lang.IllegalArgumentException
      at ...ExceptionThrowingDelegationTokenProvider.registerJob(...)
      at ...DefaultDelegationTokenManager.registerJob(DefaultDelegationTokenManager.java:617)

And moreover, we already have the failing provider in stack trace.
The rollback path has its own ERROR log too ("Failed to roll back registration of job {}").

@Savonitar Savonitar Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One addition to my note above: that holds for an Exception, but an Error slips past both catches. The realistic trigger is a NoClassDefFoundError (or other LinkageError from provider plugin code), the same failure class loadProviders already has:

} catch (Exception | NoClassDefFoundError e) {
                        // The intentional general rule is that if a provider's init method throws
                        // exception
                        // then stop the workload
                        LOG.error(
                                "Failed to initialize delegation token provider {}",
                                provider.serviceName(),
                                e);
                        throw new FlinkRuntimeException(e);
                    }

@Savonitar Savonitar Jul 7, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in e2dba9e. Both job hooks now treat Exception | LinkageError the same.

}
return registerJobMasterInternal(
jobMasterGateway,
jobId,
Expand Down Expand Up @@ -1223,6 +1237,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) {
Expand Down
34 changes: 34 additions & 0 deletions ...untime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.runtime.blob.BlobServer;
import org.apache.flink.runtime.blob.TransientBlobKey;
import org.apache.flink.runtime.blocklist.BlocklistListener;
Expand Down Expand Up @@ -63,10 +64,42 @@ public interface ResourceManagerGateway
/**
* Register a {@link JobMaster} at the resource manager.
*
* <p>Backward-compatible overload that registers without a job configuration. Equivalent to
* calling {@link #registerJobMaster(JobMasterId, ResourceID, String, JobID, Configuration,
* Duration)} with an empty configuration.
*
* @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 timeout Timeout for the future to complete
* @return Future registration response
*/
default CompletableFuture<RegistrationResponse> registerJobMaster(
JobMasterId jobMasterId,
ResourceID jobMasterResourceId,
String jobMasterAddress,
JobID jobId,
@RpcTimeout Duration timeout) {
return registerJobMaster(
jobMasterId,
jobMasterResourceId,
jobMasterAddress,
jobId,
new Configuration(),
timeout);
}

/**
* 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
*/
Expand All @@ -75,6 +108,7 @@ CompletableFuture<RegistrationResponse> registerJobMaster(
ResourceID jobMasterResourceId,
String jobMasterAddress,
JobID jobId,
Configuration jobConfiguration,
@RpcTimeout Duration timeout);

/**
Expand Down
Loading