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 8c56630692d 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.
##########
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;
+ }
+ }
+
+ @Override
+ public void unregisterJob(JobID jobId) throws Exception {
Review Comment:
Addressed in 8c56630692d133b2b079b2aa603b03b386d6c038 :
`unregisterJobInternal` now returns void and always removes the job.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java:
##########
@@ -427,6 +430,22 @@ public CompletableFuture<RegistrationResponse>
registerJobMaster(
jobMasterIdFuture,
(JobMasterGateway jobMasterGateway, JobMasterId
leadingJobMasterId) -> {
if (Objects.equals(leadingJobMasterId,
jobMasterId)) {
+ // Register with the delegation token
manager first, so a
+ // provider failure rejects the
registration and the job does
+ // not start without the tokens it
requires. LinkageError is
+ // caught so a plugin classpath failure is
reported the same
+ // way.
+ try {
+
delegationTokenManager.registerJob(jobId, jobConfiguration);
+ } catch (Exception | LinkageError e) {
Review Comment:
> Why do we want to prepare for LinkageError?
The LinkageError catch is there so a failed registration doesn't leave job
state in the providers that already registered the job.
It doesn't swallow the error: registerJob rolls back, logs, and rethrows it
unchanged. Since 8c56630692d133b2b079b2aa603b03b386d6c038 the manager also
doesn't track a job whose registration failed.
I also conducted experiments with a real JobManager JVM (session cluster,
both without HA and with ZooKeeper HA) using a provider whose registerJob()
throws NoClassDefFoundError, with and without the catch.
I see three options:
1. Keep the catch (current branch): a failed registration is rolled back on
all providers and logged at ERROR with the job and provider, and the error is
rethrown so the registration is rejected.
2. Remove the catch (the suggestion, if I understand it correctly): the
registration is rejected the same way and **the JobManager _stays up_**, but
nothing rolls back. Providers that already registered the job **keep its
state** until the job ends and the job timeout fires (HA), or until the
JobManager shuts down (no HA, or a job that keeps restarting). The manager no
longer logs the failure at ERROR. A process-level failure happens only if a
provider's token obtain later throws an Error because of that state (a crash
loop under HA).
3. Remove the catch and fail explicitly: escalate a LinkageError from
registerJob() with onFatalError, so any broken provider deployment fails the
JobManager, at the cost of the other jobs on a session cluster.
Please correct me if I'm missing something and appreciate if you can share
your opinion on these tradeoffs.
--
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]
