This is an automated email from the ASF dual-hosted git repository.
gaborgsomogyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 6afaa60978c [FLINK-40008][security] Replace fixed delegation token
renewal retry backoff with exponential backoff
6afaa60978c is described below
commit 6afaa60978cf5d1044caff7557e24f27e1fa6d18
Author: Gabor Somogyi <[email protected]>
AuthorDate: Tue Jun 30 15:22:13 2026 +0200
[FLINK-40008][security] Replace fixed delegation token renewal retry
backoff with exponential backoff
---
.../generated/security_configuration.html | 12 +++-
.../security_delegation_token_section.html | 12 +++-
.../flink/configuration/SecurityOptions.java | 26 +++++---
.../token/DefaultDelegationTokenManager.java | 56 +++++++++++++++---
.../token/DefaultDelegationTokenManagerTest.java | 69 ++++++++++++++++++++++
5 files changed, 153 insertions(+), 22 deletions(-)
diff --git a/docs/layouts/shortcodes/generated/security_configuration.html
b/docs/layouts/shortcodes/generated/security_configuration.html
index fdf554c6670..d4fc20d1981 100644
--- a/docs/layouts/shortcodes/generated/security_configuration.html
+++ b/docs/layouts/shortcodes/generated/security_configuration.html
@@ -27,10 +27,16 @@
<td>Indicates whether to start delegation tokens system for
external services.</td>
</tr>
<tr>
- <td><h5>security.delegation.tokens.renewal.retry.backoff</h5></td>
- <td style="word-wrap: break-word;">1 h</td>
+
<td><h5>security.delegation.tokens.renewal.retry.initial.backoff</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
<td>Duration</td>
- <td>The time period how long to wait before retrying to obtain new
delegation tokens after a failure.</td>
+ <td>Initial backoff time when retrying to obtain delegation tokens
after a failure. Doubles on each subsequent failure up to
security.delegation.tokens.renewal.retry.max.backoff.</td>
+ </tr>
+ <tr>
+
<td><h5>security.delegation.tokens.renewal.retry.max.backoff</h5></td>
+ <td style="word-wrap: break-word;">5 min</td>
+ <td>Duration</td>
+ <td>Maximum backoff time between retries when obtaining delegation
tokens fails. Retry delays start at
security.delegation.tokens.renewal.retry.initial.backoff and double on each
failure up to this cap.</td>
</tr>
<tr>
<td><h5>security.delegation.tokens.renewal.time-ratio</h5></td>
diff --git
a/docs/layouts/shortcodes/generated/security_delegation_token_section.html
b/docs/layouts/shortcodes/generated/security_delegation_token_section.html
index e81865a4d07..0aa8ab815d7 100644
--- a/docs/layouts/shortcodes/generated/security_delegation_token_section.html
+++ b/docs/layouts/shortcodes/generated/security_delegation_token_section.html
@@ -15,10 +15,16 @@
<td>Indicates whether to start delegation tokens system for
external services.</td>
</tr>
<tr>
- <td><h5>security.delegation.tokens.renewal.retry.backoff</h5></td>
- <td style="word-wrap: break-word;">1 h</td>
+
<td><h5>security.delegation.tokens.renewal.retry.initial.backoff</h5></td>
+ <td style="word-wrap: break-word;">10 s</td>
<td>Duration</td>
- <td>The time period how long to wait before retrying to obtain new
delegation tokens after a failure.</td>
+ <td>Initial backoff time when retrying to obtain delegation tokens
after a failure. Doubles on each subsequent failure up to
security.delegation.tokens.renewal.retry.max.backoff.</td>
+ </tr>
+ <tr>
+
<td><h5>security.delegation.tokens.renewal.retry.max.backoff</h5></td>
+ <td style="word-wrap: break-word;">5 min</td>
+ <td>Duration</td>
+ <td>Maximum backoff time between retries when obtaining delegation
tokens fails. Retry delays start at
security.delegation.tokens.renewal.retry.initial.backoff and double on each
failure up to this cap.</td>
</tr>
<tr>
<td><h5>security.delegation.tokens.renewal.time-ratio</h5></td>
diff --git
a/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
b/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
index 32d8ee7a1e6..3cfdca5c596 100644
---
a/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
+++
b/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
@@ -148,30 +148,40 @@ public class SecurityOptions {
key("security.delegation.tokens.enabled")
.booleanType()
.defaultValue(true)
-
.withDeprecatedKeys("security.kerberos.fetch.delegation-token")
.withDescription(
"Indicates whether to start delegation tokens
system for external services.");
@Documentation.Section(value =
Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 2)
- public static final ConfigOption<Duration>
DELEGATION_TOKENS_RENEWAL_RETRY_BACKOFF =
- key("security.delegation.tokens.renewal.retry.backoff")
+ public static final ConfigOption<Duration>
DELEGATION_TOKENS_RENEWAL_RETRY_INITIAL_BACKOFF =
+ key("security.delegation.tokens.renewal.retry.initial.backoff")
.durationType()
- .defaultValue(Duration.ofHours(1))
-
.withDeprecatedKeys("security.kerberos.tokens.renewal.retry.backoff")
+ .defaultValue(Duration.ofSeconds(10))
.withDescription(
- "The time period how long to wait before retrying
to obtain new delegation tokens after a failure.");
+ "Initial backoff time when retrying to obtain
delegation tokens after a failure. "
+ + "Doubles on each subsequent failure up
to "
+ +
"security.delegation.tokens.renewal.retry.max.backoff.");
@Documentation.Section(value =
Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 3)
+ public static final ConfigOption<Duration>
DELEGATION_TOKENS_RENEWAL_RETRY_MAX_BACKOFF =
+ key("security.delegation.tokens.renewal.retry.max.backoff")
+ .durationType()
+ .defaultValue(Duration.ofMinutes(5))
+
.withDeprecatedKeys("security.delegation.tokens.renewal.retry.backoff")
+ .withDescription(
+ "Maximum backoff time between retries when
obtaining delegation tokens fails. "
+ + "Retry delays start at
security.delegation.tokens.renewal.retry.initial.backoff "
+ + "and double on each failure up to this
cap.");
+
+ @Documentation.Section(value =
Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 4)
public static final ConfigOption<Double>
DELEGATION_TOKENS_RENEWAL_TIME_RATIO =
key("security.delegation.tokens.renewal.time-ratio")
.doubleType()
.defaultValue(0.75)
-
.withDeprecatedKeys("security.kerberos.tokens.renewal.time-ratio")
.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 = 4)
+ @Documentation.Section(value =
Documentation.Sections.SECURITY_DELEGATION_TOKEN, position = 5)
public static final ConfigOption<Boolean>
DELEGATION_TOKEN_PROVIDER_ENABLED =
key("enabled")
.booleanType()
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java
index e050ee1e643..d1abfefd866 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java
@@ -27,6 +27,7 @@ import
org.apache.flink.core.security.token.DelegationTokenProvider;
import org.apache.flink.core.security.token.DelegationTokenReceiver;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.InstantiationUtil;
+import org.apache.flink.util.TimeUtils;
import org.apache.flink.util.concurrent.ScheduledExecutor;
import org.slf4j.Logger;
@@ -36,6 +37,7 @@ import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;
import java.time.Clock;
+import java.time.Duration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
@@ -44,11 +46,13 @@ import java.util.ServiceLoader;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Stream;
-import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKENS_RENEWAL_RETRY_BACKOFF;
+import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKENS_RENEWAL_RETRY_INITIAL_BACKOFF;
+import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKENS_RENEWAL_RETRY_MAX_BACKOFF;
import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKENS_RENEWAL_TIME_RATIO;
import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKEN_PROVIDER_ENABLED;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -80,7 +84,13 @@ public class DefaultDelegationTokenManager implements
DelegationTokenManager {
private final double tokensRenewalTimeRatio;
- private final long renewalRetryBackoffPeriod;
+ private final long renewalRetryInitialBackoff;
+
+ private final long renewalRetryMaxBackoff;
+
+ @VisibleForTesting long currentRetryBackoff;
+
+ @VisibleForTesting long lastKnownNextRenewal = Long.MAX_VALUE;
@VisibleForTesting final Map<String, DelegationTokenProvider>
delegationTokenProviders;
@@ -106,8 +116,11 @@ public class DefaultDelegationTokenManager implements
DelegationTokenManager {
this.configuration = checkNotNull(configuration, "Flink configuration
must not be null");
this.pluginManager = pluginManager;
this.tokensRenewalTimeRatio =
configuration.get(DELEGATION_TOKENS_RENEWAL_TIME_RATIO);
- this.renewalRetryBackoffPeriod =
-
configuration.get(DELEGATION_TOKENS_RENEWAL_RETRY_BACKOFF).toMillis();
+ this.renewalRetryInitialBackoff =
+
configuration.get(DELEGATION_TOKENS_RENEWAL_RETRY_INITIAL_BACKOFF).toMillis();
+ this.renewalRetryMaxBackoff =
+
configuration.get(DELEGATION_TOKENS_RENEWAL_RETRY_MAX_BACKOFF).toMillis();
+ this.currentRetryBackoff = renewalRetryInitialBackoff;
this.delegationTokenProviders = loadProviders();
this.delegationTokenReceiverRepository =
new DelegationTokenReceiverRepository(configuration,
pluginManager);
@@ -321,6 +334,8 @@ public class DefaultDelegationTokenManager implements
DelegationTokenManager {
}
if (nextRenewal.isPresent()) {
+ lastKnownNextRenewal = nextRenewal.get();
+ currentRetryBackoff = renewalRetryInitialBackoff;
long renewalDelay =
calculateRenewalDelay(Clock.systemDefaultZone(),
nextRenewal.get());
synchronized (tokensUpdateFutureLock) {
@@ -330,7 +345,9 @@ public class DefaultDelegationTokenManager implements
DelegationTokenManager {
renewalDelay,
TimeUnit.MILLISECONDS);
}
- LOG.info("Tokens update task started with {} ms delay",
renewalDelay);
+ LOG.info(
+ "Tokens update task started with {} delay",
+
TimeUtils.formatWithHighestUnit(Duration.ofMillis(renewalDelay)));
} else {
LOG.warn(
"Tokens update task not started because either no
tokens obtained or none of the tokens specified its renewal date");
@@ -339,16 +356,17 @@ public class DefaultDelegationTokenManager implements
DelegationTokenManager {
// Ignore, may happen if shutting down.
LOG.debug("Interrupted", e);
} catch (Exception e) {
+ long delay = calculateRetryDelay(Clock.systemDefaultZone());
synchronized (tokensUpdateFutureLock) {
tokensUpdateFuture =
scheduledExecutor.schedule(
() ->
ioExecutor.execute(this::startTokensUpdate),
- renewalRetryBackoffPeriod,
+ delay,
TimeUnit.MILLISECONDS);
}
LOG.warn(
- "Failed to update tokens, will try again in {} ms",
- renewalRetryBackoffPeriod,
+ "Failed to update tokens, will try again in {}",
+ TimeUtils.formatWithHighestUnit(Duration.ofMillis(delay)),
e);
}
}
@@ -363,6 +381,28 @@ public class DefaultDelegationTokenManager implements
DelegationTokenManager {
}
}
+ @VisibleForTesting
+ long calculateRetryDelay(Clock clock) {
+ long nowMillis = clock.millis();
+ long effectiveMax;
+ if (lastKnownNextRenewal != Long.MAX_VALUE) {
+ long remaining = lastKnownNextRenewal - nowMillis;
+ // If a failure occurs close to token expiry there may not be
enough time for the
+ // normal exponential backoff to complete before the token becomes
invalid. Capping
+ // each retry delay to one third of the remaining valid window
ensures a retry always
+ // happens while the token is still live, regardless of how close
to expiry the
+ // failure occurred.
+ effectiveMax = remaining > 0 ? remaining / 3 :
renewalRetryMaxBackoff;
+ } else {
+ effectiveMax = renewalRetryMaxBackoff;
+ }
+ long base = Math.min(currentRetryBackoff, effectiveMax);
+ long jitter = (long) ((ThreadLocalRandom.current().nextDouble() - 0.5)
* base);
+ long delay = Math.max(0, Math.min(base + jitter, effectiveMax));
+ currentRetryBackoff = Math.min(currentRetryBackoff * 2,
renewalRetryMaxBackoff);
+ return delay;
+ }
+
@VisibleForTesting
long calculateRenewalDelay(Clock clock, long nextRenewal) {
long now = clock.millis();
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManagerTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManagerTest.java
index c7a4cf6f958..5f78c79a89d 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManagerTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManagerTest.java
@@ -29,6 +29,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.time.Clock;
+import java.time.Duration;
import java.time.ZoneId;
import java.util.Collections;
import java.util.HashMap;
@@ -39,6 +40,8 @@ import java.util.concurrent.atomic.AtomicInteger;
import static java.time.Instant.ofEpochMilli;
import static
org.apache.flink.configuration.ConfigurationUtils.getBooleanConfigOption;
+import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKENS_RENEWAL_RETRY_INITIAL_BACKOFF;
+import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKENS_RENEWAL_RETRY_MAX_BACKOFF;
import static
org.apache.flink.configuration.SecurityOptions.DELEGATION_TOKENS_RENEWAL_TIME_RATIO;
import static
org.apache.flink.core.security.token.DelegationTokenProvider.CONFIG_PREFIX;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -231,4 +234,70 @@ public class DefaultDelegationTokenManagerTest {
Clock constantClock = Clock.fixed(ofEpochMilli(100),
ZoneId.systemDefault());
assertEquals(50,
delegationTokenManager.calculateRenewalDelay(constantClock, 200));
}
+
+ @Test
+ public void calculateRetryDelayShouldDoubleOnConsecutiveFailures() {
+ long initialMs = Duration.ofSeconds(10).toMillis();
+ Configuration configuration = new Configuration();
+ configuration.set(getBooleanConfigOption(CONFIG_PREFIX +
".throw.enabled"), false);
+ configuration.set(DELEGATION_TOKENS_RENEWAL_RETRY_INITIAL_BACKOFF,
Duration.ofSeconds(10));
+ configuration.set(DELEGATION_TOKENS_RENEWAL_RETRY_MAX_BACKOFF,
Duration.ofMinutes(5));
+ DefaultDelegationTokenManager manager =
+ new DefaultDelegationTokenManager(configuration, null, null,
null);
+
+ Clock clock = Clock.fixed(ofEpochMilli(0), ZoneId.systemDefault());
+ long delay1 = manager.calculateRetryDelay(clock);
+ long delay2 = manager.calculateRetryDelay(clock);
+ long delay3 = manager.calculateRetryDelay(clock);
+
+ // Each delay should be within [0, 2 * initial * 2^(n-1)] accounting
for ±50% jitter.
+ assertTrue(delay1 >= 0 && delay1 <= initialMs * 2);
+ assertTrue(delay2 >= 0 && delay2 <= initialMs * 4);
+ assertTrue(delay3 >= 0 && delay3 <= initialMs * 8);
+ // The base must have doubled: currentRetryBackoff after 3 calls is
min(80s, 5min) = 80s.
+ assertEquals(Duration.ofSeconds(80).toMillis(),
manager.currentRetryBackoff);
+ }
+
+ @Test
+ public void calculateRetryDelayShouldResetAfterSuccess() {
+ long initialMs = Duration.ofSeconds(10).toMillis();
+ Configuration configuration = new Configuration();
+ configuration.set(getBooleanConfigOption(CONFIG_PREFIX +
".throw.enabled"), false);
+ configuration.set(DELEGATION_TOKENS_RENEWAL_RETRY_INITIAL_BACKOFF,
Duration.ofSeconds(10));
+ configuration.set(DELEGATION_TOKENS_RENEWAL_RETRY_MAX_BACKOFF,
Duration.ofMinutes(5));
+ DefaultDelegationTokenManager manager =
+ new DefaultDelegationTokenManager(configuration, null, null,
null);
+
+ // Ramp up the backoff via two failures.
+ Clock clock = Clock.fixed(ofEpochMilli(0), ZoneId.systemDefault());
+ manager.calculateRetryDelay(clock);
+ manager.calculateRetryDelay(clock);
+ // Simulate success: reset currentRetryBackoff (as startTokensUpdate()
would).
+ manager.currentRetryBackoff = initialMs;
+
+ long delayAfterReset = manager.calculateRetryDelay(clock);
+ assertTrue(delayAfterReset >= 0 && delayAfterReset <= initialMs * 2);
+ assertEquals(initialMs * 2, manager.currentRetryBackoff);
+ }
+
+ @Test
+ public void calculateRetryDelayShouldCapToTtlBound() {
+ Configuration configuration = new Configuration();
+ configuration.set(getBooleanConfigOption(CONFIG_PREFIX +
".throw.enabled"), false);
+ configuration.set(DELEGATION_TOKENS_RENEWAL_RETRY_INITIAL_BACKOFF,
Duration.ofSeconds(10));
+ configuration.set(DELEGATION_TOKENS_RENEWAL_RETRY_MAX_BACKOFF,
Duration.ofMinutes(5));
+ DefaultDelegationTokenManager manager =
+ new DefaultDelegationTokenManager(configuration, null, null,
null);
+
+ // Simulate a failure close to token expiry (30 s remaining). The
delay must be capped
+ // so that the retry happens while the token is still valid (at most
30 s / 3 = 10 s).
+ Clock clock = Clock.fixed(ofEpochMilli(0), ZoneId.systemDefault());
+ manager.lastKnownNextRenewal = Duration.ofSeconds(30).toMillis();
+
+ long delay = manager.calculateRetryDelay(clock);
+
+ // Delay must not exceed the TTL cap (30 s / 3 = 10 s), with jitter
the max is 10 s.
+ assertTrue(delay <= Duration.ofSeconds(10).toMillis());
+ assertTrue(delay >= 0);
+ }
}
