Savonitar commented on code in PR #28639:
URL: https://github.com/apache/flink/pull/28639#discussion_r3539498842
##########
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);
+ }
+ }
+
+ @Override
+ public void registerJob(JobID jobId, Configuration jobConfiguration)
throws Exception {
+ try {
+ for (DelegationTokenProvider provider :
delegationTokenProviders.values()) {
+ provider.registerJob(jobId, jobConfiguration);
+ }
+ } catch (Exception e) {
+ // If any of the providers fail to register, then unregister the
job from them all.
+ // unregisterJob is idempotent, so it is safe to call it for
providers that were never
+ // (or only partially) registered for this job before the failure.
The rollback must
+ // never mask the original failure, so swallow any rollback
exception.
+ try {
+ unregisterJob(jobId);
+ } catch (Exception rollbackException) {
+ LOG.error("Failed to roll back registration of job {}", jobId,
rollbackException);
+ }
+ LOG.error("Failed to register job {}", jobId, e);
+ throw e;
+ }
+ }
+
+ @Override
+ public void unregisterJob(JobID jobId) throws Exception {
+ for (DelegationTokenProvider provider :
delegationTokenProviders.values()) {
+ try {
+ provider.unregisterJob(jobId);
+ } catch (Exception e) {
+ LOG.error("Failed to unregister job for provider {}",
provider.serviceName(), e);
Review Comment:
Good catch, added in dfb3cd6
--
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]
