Savonitar commented on code in PR #28639:
URL: https://github.com/apache/flink/pull/28639#discussion_r4064026722
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -416,13 +663,191 @@ long calculateRenewalDelay(Clock clock, long
nextRenewal) {
return renewalDelay;
}
- /** Stops re-occurring token obtain task. */
+ /**
+ * Stops the re-occurring token obtain task, releases the listener, and
unregisters the jobs of
+ * the ending session. Providers stay usable for a later {@link
#start(Listener)}. Their
+ * teardown happens in {@link #close()}.
+ */
@Override
public void stop() {
LOG.info("Stopping credential renewal");
- stopTokensUpdate();
+ synchronized (tokensUpdateFutureLock) {
+ // Mark not running, cancel the pending cycle, and reset the
re-obtain bookkeeping
+ // atomically, so a re-obtain racing shutdown cannot schedule a
cycle for a manager
+ // that is shutting down.
+ running = false;
+ stopTokensUpdate();
+ reobtainScheduled = false;
+ lastReobtainAtMillis = NO_PREVIOUS_REOBTAIN;
+ // Release the listener: keeping it would pin the disposed
ResourceManager of a
+ // revoked leadership session, forever on a standby that never
regains leadership.
+ listener = null;
+ }
+
+ // Unregister all jobs: running jobs re-register with the next
session, ended jobs never
+ // would and their entries would leak in the providers.
+ for (JobID jobId : registeredJobs) {
+ try {
+ unregisterJobInternal(jobId);
+ } catch (Exception | LinkageError e) {
+ // Guards the cleanup against pathological errors from a
broken plugin's
+ // serviceName().
+ LOG.error("Failed to unregister job {} while stopping the
manager", jobId, e);
+ }
+ }
LOG.info("Stopped credential renewal");
}
+
+ /**
+ * Terminal teardown: ends any active session via {@link #stop()} and then
stops all providers,
+ * exactly once. Called by the component that created the manager at
process shutdown, not on
+ * ResourceManager leadership changes.
+ */
+ @Override
+ public void close() {
+ // Flip the flag before stopping anything. start() checks it under
+ // tokensUpdateFutureLock, so a racing start() either fails the check
or has its
+ // session ended by the stop() below (see start()). At most one obtain
may still
+ // overlap the provider stop() below, which the provider threading
contract covers.
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+ stop();
+ for (DelegationTokenProvider provider :
delegationTokenProviders.values()) {
+ try {
+ provider.stop();
+ } catch (Throwable t) {
+ LOG.error("Failed to stop delegation token provider {}",
provider.serviceName(), t);
+ }
+ }
+ }
+
+ @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), "
+ + "ignoring the request.");
+ return;
+ }
+ if (!running) {
+ LOG.debug(
+ "A re-obtain of delegation tokens was requested while
the manager is not "
+ + "running (not started yet, or already
stopped), ignoring the "
+ + "request.");
+ return;
+ }
+ // An already scheduled re-obtain that has not started yet covers
this request too.
+ 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.relativeTimeMillis();
+ long delayMillis =
+ lastReobtainAtMillis == NO_PREVIOUS_REOBTAIN
+ ? 0L
+ : Math.max(0L, lastReobtainAtMillis +
reobtainCooldownMillis - now);
+ // Only bring the next cycle forward, never push a pending cycle
later, or a
+ // short-lived token could expire before it is renewed. The
nextScheduledAtMillis >
+ // now guard skips an already-fired future, so this never bypasses
the cooldown.
+ if (tokensUpdateFuture != null
+ && nextScheduledAtMillis > now
+ && nextScheduledAtMillis - now < delayMillis) {
+ delayMillis = nextScheduledAtMillis - now;
+ }
+ // Anchor the cooldown to when the cycle will run, not to this
request, so a request
+ // arriving right after a deferred cycle fired cannot run a second
cycle back to back.
+ lastReobtainAtMillis = now + delayMillis;
+ 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 {
+ // Hand providers a copy so plugin code cannot mutate the caller's
live job configuration.
+ // clone() locks the backing map. Like the copy constructor, the copy
is shallow.
+ final Configuration providerJobConfiguration =
jobConfiguration.clone();
+ final boolean previouslyRegistered = registeredJobs.contains(jobId);
+ DelegationTokenProvider failedProvider = null;
+ try {
+ for (DelegationTokenProvider provider :
delegationTokenProviders.values()) {
+ failedProvider = provider;
+ provider.registerJob(jobId, providerJobConfiguration);
+ }
+ registeredJobs.add(jobId);
+ } catch (Exception | LinkageError e) {
+ // LinkageError is included because provider plugin code can fail
class resolution.
+ if (previouslyRegistered) {
+ // A failed re-registration must not roll back: the job
registered successfully
+ // before and its tasks may still be running.
+ LOG.error(
+ "Failed to re-register job {} for provider {}, keeping
the previous "
+ + "registration",
+ jobId,
+ failedProvider == null ? "<none>" :
failedProvider.serviceName(),
+ e);
+ } else {
+ // First registration: roll back from all providers
(unregisterJob is idempotent).
+ // The rollback must never mask the original failure.
+ try {
+ if (!unregisterJobInternal(jobId)) {
+ // Keep the job tracked so stop() or a registration
retry can release the
+ // provider state left behind.
+ registeredJobs.add(jobId);
+ }
+ } catch (Exception | LinkageError rollbackException) {
+ LOG.error(
+ "Failed to roll back registration of job {}",
jobId, rollbackException);
+ }
+ LOG.error(
+ "Failed to register job {} for provider {}",
+ jobId,
+ failedProvider == null ? "<none>" :
failedProvider.serviceName(),
+ e);
+ }
+ throw e;
Review Comment:
Yes, the previous implementation retained jobs whose rollback or
unregistration failed. My intention was to give providers **another** cleanup
attempt in stop().
I've updated this in aa1b641aa45 to follow your suggestion: If no successful
registration is currently tracked, a failed registration triggers best-effort
rollback across all providers and leaves the job untracked. Unregistration also
removes the job from the manager even if provider cleanup fails. Failures are
logged, and the job is not retained for another cleanup attempt.
I kept `previouslyRegistered` specifically to protect an existing successful
registration. A JobMaster can re-register while its tasks are still running. If
that attempt fails, rolling back would unregister the job from every provider
and remove state those tasks still need. The set now tracks only successful
registrations, with no entries retained just for pending cleanup.
--
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]
