Savonitar commented on code in PR #28639:
URL: https://github.com/apache/flink/pull/28639#discussion_r3539676896
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -416,13 +531,114 @@ long calculateRenewalDelay(Clock clock, long
nextRenewal) {
return renewalDelay;
}
+ @VisibleForTesting
+ void setClock(Clock clock) {
+ this.clock = clock;
+ }
+
/** Stops re-occurring token obtain task. */
@Override
public void stop() {
LOG.info("Stopping credential renewal");
- stopTokensUpdate();
+ synchronized (tokensUpdateFutureLock) {
+ // Mark stopped, cancel the pending cycle, and reset on-demand
re-obtain bookkeeping
+ // atomically, so a concurrent reobtainDelegationTokens() cannot
leave a live future
+ // orphaned after stop and a later start() does not inherit stale
state.
+ stopped = true;
+ stopTokensUpdate();
+ reobtainScheduled = false;
+ lastReobtainAtMillis = NO_PREVIOUS_REOBTAIN;
+ }
+
+ for (DelegationTokenProvider provider :
delegationTokenProviders.values()) {
+ try {
+ provider.stop();
+ } catch (Throwable t) {
+ LOG.error("Failed to stop delegation token provider {}",
provider.serviceName(), t);
+ }
+ }
LOG.info("Stopped credential renewal");
}
+
+ @Override
+ public void reobtainDelegationTokens() {
+ synchronized (tokensUpdateFutureLock) {
+ if (scheduledExecutor == null || ioExecutor == null) {
+ LOG.debug(
+ "A re-obtain of delegation tokens was requested but
the manager was "
+ + "constructed without executors (one-shot
obtain path); the "
+ + "request is ignored.");
+ return;
+ }
+ if (stopped) {
+ LOG.debug(
+ "A re-obtain of delegation tokens was requested after
the manager was "
+ + "stopped; the request is ignored.");
+ return;
+ }
+ // Dedupe: if an on-demand re-obtain is already scheduled and has
not started yet, the
+ // newly registered job(s) will be covered by it, so coalesce this
request into it.
+ if (reobtainScheduled) {
+ LOG.debug("A re-obtain of delegation tokens is already
scheduled; coalescing.");
+ return;
+ }
+ // Cooldown: bound how often on-demand re-obtains can run by
deferring this cycle until
+ // at least reobtainCooldownMillis have passed since the previous
on-demand re-obtain.
+ long now = clock.millis();
+ long delayMillis =
+ lastReobtainAtMillis == NO_PREVIOUS_REOBTAIN
+ ? 0L
+ : Math.max(0L, lastReobtainAtMillis +
reobtainCooldownMillis - now);
+ // Only bring the next cycle forward: if a cycle (e.g. the
periodic renewal) is still
+ // pending and scheduled to fire sooner than the cooldown-deferred
time, fire at that
+ // earlier time instead of pushing it later — otherwise a
short-lived token could expire
+ // before it is renewed. The nextScheduledAtMillis > now guard
skips an already-fired
+ // future that has not yet been re-armed, so this never bypasses
the cooldown.
+ if (tokensUpdateFuture != null
+ && nextScheduledAtMillis > now
+ && nextScheduledAtMillis - now < delayMillis) {
+ delayMillis = nextScheduledAtMillis - now;
+ }
+ lastReobtainAtMillis = now;
+ reobtainScheduled = true;
+ LOG.debug(
+ "Re-obtain of delegation tokens requested; scheduling an
obtain cycle in {}",
+
TimeUtils.formatWithHighestUnit(Duration.ofMillis(delayMillis)));
+ scheduleRenewalLocked(delayMillis);
Review Comment:
Yes, exactly right. Fixed in 55b0566: any throwable from schedule() now
resets reobtainScheduled and nextScheduledAtMillis (same bookkeeping as the
RejectedExecutionException branch) and is rethrown.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
