gaborgsomogyi commented on code in PR #28639:
URL: https://github.com/apache/flink/pull/28639#discussion_r3704775145
##########
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`? Giving a meaningful rejection
is fine here but as a general saying any provider that throws such should not
be considered healthy. To be exact I'm against to treat providers inside the
manager which throw such exception to be tracked. Temporary exceptions can
happen but this is deployment/compile issue which should just block workloads.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java:
##########
@@ -63,10 +64,42 @@ public interface ResourceManagerGateway
/**
* Register a {@link JobMaster} at the resource manager.
*
+ * <p>Backward-compatible overload that registers without a job
configuration. Equivalent to
+ * calling {@link #registerJobMaster(JobMasterId, ResourceID, String,
JobID, Configuration,
+ * Duration)} with an empty configuration.
+ *
+ * @param jobMasterId The fencing token for the JobMaster leader
+ * @param jobMasterResourceId The resource ID of the JobMaster that
registers
+ * @param jobMasterAddress The address of the JobMaster that registers
+ * @param jobId The Job ID of the JobMaster that registers
+ * @param timeout Timeout for the future to complete
+ * @return Future registration response
+ */
+ default CompletableFuture<RegistrationResponse> registerJobMaster(
Review Comment:
IIUC it's a dead weight that only exists to avoid touching two test files.
If that's true maybe we can modify those tests
##########
flink-runtime/src/test/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManagerTest.java:
##########
@@ -211,8 +231,9 @@ void startTokensUpdate() {
}
};
- delegationTokenManager.startTokensUpdate();
+ // The first two cycles fail and schedule a retry each. The third
succeeds.
ExceptionThrowingDelegationTokenProvider.throwInUsage.set(true);
+ delegationTokenManager.start(tokens -> {});
Review Comment:
We've migrated from `startTokensUpdate` to `start` which is fine but then it
worth to be symmetric and use `stop` instead of `stopTokensUpdate`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -92,6 +101,22 @@ public class DefaultDelegationTokenManager implements
DelegationTokenManager {
@VisibleForTesting long lastKnownNextRenewal = Long.MAX_VALUE;
+ private final long reobtainCooldownMillis;
+
+ /**
+ * Clock used for renewal and cooldown timing. Renewal math reads absolute
time (a token's
+ * validUntil is an absolute epoch), while scheduling and the cooldown
read relative time, which
+ * wall-clock adjustments cannot distort. Never mix the two in one
expression.
+ */
+ private final Clock clock;
+
+ /**
+ * Serializes the obtain-and-broadcast cycle so that, even though {@code
cancel(true)} does not
+ * wait for an in-flight cycle and the IO executor is multi-threaded, two
cycles can never run
+ * concurrently and broadcast tokens out of order.
+ */
+ private final Object obtainLock = new Object();
Review Comment:
Two locks here, each protecting a distinct concern: `obtainLock` serializes
one full obtain->broadcast->retry-bookkeeping cycle, `tokensUpdateFutureLock`
guards scheduling/lifecycle state (when the next cycle runs, `running`,
`sessionEpoch`, `listener`). `tokensUpdateFutureLock` doesn't reflect that, it
sounds like it only guards one field. Suggest renaming `obtainLock` to
`renewalCycleLock` and `tokensUpdateFutureLock` to `schedulingLock`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -307,67 +411,209 @@ protected Optional<Long>
obtainDelegationTokensAndGetNextRenewal(
public void start(Listener listener) throws Exception {
checkNotNull(scheduledExecutor, "Scheduled executor must not be null");
checkNotNull(ioExecutor, "IO executor must not be null");
- this.listener = checkNotNull(listener, "Listener must not be null");
+ checkNotNull(listener, "Listener must not be null");
synchronized (tokensUpdateFutureLock) {
- checkState(tokensUpdateFuture == null, "Manager is already
started");
+ // Checked under the lock so a start() arriving after close()
fails instead of
+ // resurrecting a session against stopped providers. A start()
racing close() can
+ // still slip past the check. close()'s stop() then ends its
session under this
+ // lock, so its inline first cycle either skips on running ==
false or runs at most
+ // one obtain that cannot deliver or reschedule and may overlap
the provider stop()
+ // (see close()).
+ checkState(
+ !closed.get(),
+ "The delegation token manager is already closed, its
providers are stopped");
+ if (running) {
+ LOG.warn("DelegationTokenManager is already started, ignoring
redundant start()");
+ return;
+ }
+ this.listener = listener;
+ sessionEpoch++;
+ // Set before the inline first cycle below: startTokensUpdate() and
+ // maybeScheduleRenewal() gate on it.
+ running = true;
+ }
+
+ // A new session must not inherit the previous session's retry backoff
or renewal
+ // deadline. obtainLock orders this reset after any still-running
previous cycle. Not
+ // nested in the block above to keep the obtainLock ->
tokensUpdateFutureLock order.
+ synchronized (obtainLock) {
+ currentRetryBackoff = renewalRetryInitialBackoff;
+ lastKnownNextRenewal = Long.MAX_VALUE;
}
startTokensUpdate();
}
@VisibleForTesting
void startTokensUpdate() {
- try {
- LOG.info("Starting tokens update task");
- DelegationTokenContainer container = new
DelegationTokenContainer();
- Optional<Long> nextRenewal =
obtainDelegationTokensAndGetNextRenewal(container);
-
- if (container.hasTokens()) {
-
delegationTokenReceiverRepository.onNewTokensObtained(container);
-
- LOG.info("Notifying listener about new tokens");
- checkNotNull(listener, "Listener must not be null");
-
listener.onNewTokensObtained(InstantiationUtil.serializeObject(container));
- LOG.info("Listener notified successfully");
- } else {
- LOG.warn("No tokens obtained so skipping notifications");
+ final long cycleEpoch;
+ synchronized (tokensUpdateFutureLock) {
+ // Clear the dedupe flag so later on-demand requests can schedule
a fresh cycle.
+ reobtainScheduled = false;
+ // Stopped or never started: skip the cycle. The providers may
already be stopped
+ // and the listener may not be set yet.
+ if (!running) {
+ return;
}
+ cycleEpoch = sessionEpoch;
+ }
+ // Serialize the obtain-and-broadcast so a re-obtain racing the
periodic renewal cannot run
+ // two cycles concurrently on the (multi-threaded) IO executor and
broadcast out of order.
+ synchronized (obtainLock) {
+ try {
+ LOG.info("Starting tokens update task");
+ DelegationTokenContainer container = new
DelegationTokenContainer();
+ Optional<Long> nextRenewal =
obtainDelegationTokensAndGetNextRenewal(container);
+
+ if (container.hasTokens()) {
+ // stop() does not wait for an in-flight cycle: re-check
running so a resumed
+ // cycle does not notify the stopped session's listener,
and compare epochs
+ // so a cycle begun under an earlier session cannot
deliver into the next
+ // one (see sessionEpoch). A stop() right after this read
still lets one
+ // delivery through, which is benign.
+ final Listener currentListener;
+ synchronized (tokensUpdateFutureLock) {
+ currentListener = running && cycleEpoch ==
sessionEpoch ? listener : null;
+ }
+ if (currentListener != null) {
+
delegationTokenReceiverRepository.onNewTokensObtained(container);
+
+ LOG.info("Notifying listener about new tokens");
+ currentListener.onNewTokensObtained(
+ InstantiationUtil.serializeObject(container));
+ LOG.info("Listener notified successfully");
+ } else {
+ LOG.info(
+ "Manager stopped while the tokens were being
obtained, skipping "
+ + "notifications");
+ }
+ } else {
+ LOG.warn("No tokens obtained so skipping notifications");
+ }
- if (nextRenewal.isPresent()) {
- lastKnownNextRenewal = nextRenewal.get();
- currentRetryBackoff = renewalRetryInitialBackoff;
- long renewalDelay =
- calculateRenewalDelay(Clock.systemDefaultZone(),
nextRenewal.get());
- synchronized (tokensUpdateFutureLock) {
- tokensUpdateFuture =
- scheduledExecutor.schedule(
- () ->
ioExecutor.execute(this::startTokensUpdate),
- renewalDelay,
- TimeUnit.MILLISECONDS);
+ if (nextRenewal.isPresent()) {
+ lastKnownNextRenewal = nextRenewal.get();
+ currentRetryBackoff = renewalRetryInitialBackoff;
+ long renewalDelay = calculateRenewalDelay(clock,
nextRenewal.get());
+ long effectiveDelay = maybeScheduleRenewal(renewalDelay);
+ if (effectiveDelay >= 0) {
+ LOG.info(
+ "Tokens update task started with {} delay",
+
TimeUtils.formatWithHighestUnit(Duration.ofMillis(effectiveDelay)));
+ } else {
+ LOG.info("Tokens update task not rescheduled, the
manager is not running");
+ }
+ } else {
+ LOG.warn(
+ "Tokens update task not started because either no
tokens obtained or none of the tokens specified its renewal date");
}
- 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");
+ } catch (InterruptedException e) {
+ // Ignore, may happen if shutting down.
+ LOG.debug("Interrupted", e);
+ } catch (Exception e) {
+ long delay = calculateRetryDelay(clock);
+ long effectiveDelay;
+ try {
+ effectiveDelay = maybeScheduleRenewal(delay);
+ } catch (Throwable schedulingFailure) {
+ // The original failure was not logged yet, keep it
attached.
+ schedulingFailure.addSuppressed(e);
+ throw schedulingFailure;
+ }
+ if (effectiveDelay >= 0) {
+ LOG.warn(
+ "Failed to update tokens, will try again in {}",
+
TimeUtils.formatWithHighestUnit(Duration.ofMillis(effectiveDelay)),
+ e);
+ } else {
+ LOG.warn(
+ "Failed to update tokens, no retry scheduled
because the manager is "
+ + "not running",
+ e);
+ }
+ }
+ }
+ }
+
+ /**
+ * Schedules a one-shot token-obtain-and-broadcast cycle after {@code
delayMs}, replacing any
+ * pending renewal. A delay of {@code 0} brings the next cycle forward to
now. Must only be
+ * called after {@link #start(Listener)} (the scheduled and IO executors
are non-null then) and
+ * while holding {@link #tokensUpdateFutureLock}.
+ */
+ @GuardedBy("tokensUpdateFutureLock")
+ private void scheduleRenewalLocked(long delayMs) {
+ stopTokensUpdate();
+ nextScheduledAtMillis = clock.relativeTimeMillis() + delayMs;
+ try {
+ tokensUpdateFuture =
+ scheduledExecutor.schedule(
+ () -> {
+ try {
+
ioExecutor.execute(this::startTokensUpdate);
+ } catch (RejectedExecutionException e) {
+ // IO executor is shutting down: drop the
cycle but release the
+ // dedupe flag so it cannot get stuck if
the manager is reused.
+ synchronized (tokensUpdateFutureLock) {
+ reobtainScheduled = false;
+ }
+ LOG.debug("Tokens update task rejected by
IO executor", e);
+ }
+ },
+ delayMs,
+ TimeUnit.MILLISECONDS);
Review Comment:
`RejectedExecutionException` doesn't necessarily mean shutdown, it's also
thrown when a bounded queue is saturated while the executor is still alive.
`scheduledExecutor` is a generic `ScheduledExecutor`, so we can't assume
shutdown here. If it's actually saturation, this catch drops the renewal cycle
for the rest of the session with no retry. Shouldn't this schedule a retry
instead of assuming shutdown?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerJobMasterTest.java:
##########
@@ -152,6 +161,39 @@ void testRegisterJobMaster() {
.isInstanceOf(JobMasterRegistrationSuccess.class);
}
+ /**
+ * FLIP-588: if the delegation token manager rejects the job (its {@code
registerJob} throws),
+ * the ResourceManager must reject the JobMaster registration so the job
does not start without
+ * the tokens it requires. This also exercises the widened (6-arg) {@code
registerJobMaster} RPC
+ * that carries the job {@link Configuration}.
+ */
+ @Test
+ void testRegisterJobMasterRejectedWhenDelegationTokenRegistrationFails()
throws Exception {
+ // Rebuild the RM service with a delegation token manager that rejects
registerJob.
+ resourceManagerService.rethrowFatalErrorIfAny();
+ resourceManagerService.cleanUp();
+ final FlinkRuntimeException failure =
+ new FlinkRuntimeException("registerJob rejected by provider");
+ createAndStartResourceManagerService(new
RejectingDelegationTokenManager(failure));
+
+ final CompletableFuture<RegistrationResponse> registrationFuture =
+ resourceManagerGateway.registerJobMaster(
+ jobMasterGateway.getFencingToken(),
+ jobMasterResourceId,
+ jobMasterGateway.getAddress(),
+ jobId,
+ new Configuration(),
+ TIMEOUT);
+
+ final RegistrationResponse response =
+ registrationFuture.get(TIMEOUT.toMillis(),
TimeUnit.MILLISECONDS);
+ assertThat(response).isInstanceOf(RegistrationResponse.Failure.class);
+ final Throwable reason = ((RegistrationResponse.Failure)
response).getReason();
+ assertThat(reason.getMessage()).contains(jobId.toString());
+ assertThat(reason.getMessage()).contains("delegation token manager");
+ assertThat(reason.getCause().getMessage()).contains("registerJob
rejected by provider");
Review Comment:
Nice that this locks in the RPC-level failure response, but it doesn't cover
what happens to `jobLeaderIdService` afterward.
`jobLeaderIdService.addJob(jobId)` runs unconditionally before the
delegation-token check and nothing removes it on this failure path, so the job
stays tracked until either a retry succeeds (`containsJob` short-circuits
`addJob` and the retry re-attempts `registerJob`) or the leader-id timeout
fires. Could we add a test that retries registration after this failure and
asserts it eventually succeeds end-to-end?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -377,13 +623,14 @@ void stopTokensUpdate() {
if (tokensUpdateFuture != null) {
tokensUpdateFuture.cancel(true);
tokensUpdateFuture = null;
+ nextScheduledAtMillis = Long.MAX_VALUE;
}
}
}
@VisibleForTesting
Review Comment:
`currentRetryBackoff` and `lastKnownNextRenewal` are only ever read/written
while `obtainLock` is held (`start()`, `startTokensUpdate()`, here), but
neither is annotated. Should both be `@GuardedBy("obtainLock")` for consistency
with the rest of the fields in this class.
##########
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:
Not sure I get why we have this catch logic. I've the feeling that we want
to hack around and be adaptive to wrongly implemented providers which throw
exception in register/unregister. IIUC the actual code is:
- Tracking half successful registrations
- When unregister was not fully successful then it remains in the manager's
list infinitely without any retry or GC
My personal philosophy is that:
- We should not track any job which has thrown exception in register. Do
unregister, ignore exceptions with error and that's it.
- If a provider throws exception in unregister then move on and drop from
manager
Maybe there is a reason to add code complexity like `previouslyRegistered`
or similar but not yet get it. My general saying is that there must be a good
reason to track partially successful registration because it can degrade
stability and hard to debug.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -139,12 +240,15 @@ public DefaultDelegationTokenManager(
private Map<String, DelegationTokenProvider> loadProviders() {
LOG.info("Loading delegation token providers");
+ // Handed to every provider so it can request an immediate re-obtain
later, from any
+ // thread, decoupled from the registerJob call stack.
+ final DelegationTokenManagerCallback callback =
this::reobtainDelegationTokens;
Map<String, DelegationTokenProvider> providers = new HashMap<>();
Consumer<DelegationTokenProvider> loadProvider =
(provider) -> {
try {
if (isProviderEnabled(configuration,
provider.serviceName())) {
- provider.init(configuration);
+ provider.init(configuration, callback);
Review Comment:
We can just hardcode it, right?
##########
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);
Review Comment:
To clarify my earlier comment on the test: I'm not claiming a prod bug here,
this looks self-healing by design (retry skips `addJob` via `containsJob`,
cleanup happens via `removeJob`/leader-id timeout otherwise). What's missing is
bookkeeping test coverage: a test that fails delegation-token registration
once, then retries, and asserts the retry completes end-to-end without leaving
`jobLeaderIdService` in a duplicated or stale state. This path touches token
delivery so I would like it locked in by a test rather than relying on
inspection.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DelegationTokenManager.java:
##########
@@ -61,4 +63,37 @@ interface Listener {
/** Stops re-occurring token obtain task. */
void stop();
+
+ /**
+ * Requests an immediate, asynchronous token-obtain-and-distribute cycle,
bringing the next
+ * cycle forward instead of waiting for the periodic renewal. May be
called from any thread;
+ * it is a no-op on a manager constructed without executors (the one-shot
obtain path).
+ * Concurrent requests are coalesced and a configurable cooldown may
apply, so a call does
+ * not necessarily map to exactly one obtain.
+ *
+ * <p>Backs {@link
+ *
org.apache.flink.core.security.token.DelegationTokenManagerCallback#reobtainDelegationTokens()}.
+ */
+ default void reobtainDelegationTokens() {}
Review Comment:
I've not seen any such hardcode user who has done that but we can keep this
to be on the safe side 🙂
##########
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:
Having a function which is not doing what it actually tells to do is just
bad. `unregisterJobInternal` removes jobId under some circumstances. We must
either unregister or throw exception but not silent ignore the function intent.
##########
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()) {
Review Comment:
I think previously I've not made distinction between `stop` and `close`. Let
me explain my current understanding and correct me if I'm wrong. `stop`
(together with `start`) can happen when HA kicks in during failover and it
makes sure that new leader re-obtained tokens. `stop` has nothing to do with
freeing resources allocated by the provider and this function called regularly.
`close` happens during process shutdown and practically a single no way back
action which should free resources. At least this is what I see from the
manager from naming perspective. If this is true then maybe we can call
`DelegationTokenProvider.stop` as `DelegationTokenProvider.close` just to have
a single convention. If my understanding is correct then this is a one way
process shutdown and good as-is and we shouldn't care about any race.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/security/token/DefaultDelegationTokenManager.java:
##########
@@ -310,64 +361,127 @@ public void start(Listener listener) throws Exception {
this.listener = checkNotNull(listener, "Listener must not be null");
synchronized (tokensUpdateFutureLock) {
checkState(tokensUpdateFuture == null, "Manager is already
started");
+ stopped = false;
}
startTokensUpdate();
}
@VisibleForTesting
void startTokensUpdate() {
- try {
- LOG.info("Starting tokens update task");
- DelegationTokenContainer container = new
DelegationTokenContainer();
- Optional<Long> nextRenewal =
obtainDelegationTokensAndGetNextRenewal(container);
-
- if (container.hasTokens()) {
-
delegationTokenReceiverRepository.onNewTokensObtained(container);
-
- LOG.info("Notifying listener about new tokens");
- checkNotNull(listener, "Listener must not be null");
-
listener.onNewTokensObtained(InstantiationUtil.serializeObject(container));
- LOG.info("Listener notified successfully");
- } else {
- LOG.warn("No tokens obtained so skipping notifications");
+ synchronized (tokensUpdateFutureLock) {
+ // The obtain cycle is starting: clear the dedupe flag so later
on-demand requests can
+ // schedule a fresh cycle.
+ reobtainScheduled = false;
+ // If stop() ran before this cycle (already handed to the IO
executor) began, skip the
+ // obtain/broadcast: the providers may already be stopped. Safe
via this lock's
+ // happens-before with stop(). The dedupe flag is cleared above,
so it is never stuck.
+ if (stopped) {
+ return;
}
+ }
+ // Serialize the obtain-and-broadcast so a re-obtain racing the
periodic renewal cannot run
+ // two cycles concurrently on the (multi-threaded) IO executor and
broadcast out of order.
+ synchronized (obtainLock) {
+ try {
+ LOG.info("Starting tokens update task");
+ DelegationTokenContainer container = new
DelegationTokenContainer();
+ Optional<Long> nextRenewal =
obtainDelegationTokensAndGetNextRenewal(container);
+
+ if (container.hasTokens()) {
+
delegationTokenReceiverRepository.onNewTokensObtained(container);
+
+ LOG.info("Notifying listener about new tokens");
+ checkNotNull(listener, "Listener must not be null");
+
listener.onNewTokensObtained(InstantiationUtil.serializeObject(container));
+ LOG.info("Listener notified successfully");
+ } else {
+ LOG.warn("No tokens obtained so skipping notifications");
+ }
- if (nextRenewal.isPresent()) {
- lastKnownNextRenewal = nextRenewal.get();
- currentRetryBackoff = renewalRetryInitialBackoff;
- long renewalDelay =
- calculateRenewalDelay(Clock.systemDefaultZone(),
nextRenewal.get());
- synchronized (tokensUpdateFutureLock) {
- tokensUpdateFuture =
- scheduledExecutor.schedule(
- () ->
ioExecutor.execute(this::startTokensUpdate),
- renewalDelay,
- TimeUnit.MILLISECONDS);
+ if (nextRenewal.isPresent()) {
+ lastKnownNextRenewal = nextRenewal.get();
+ currentRetryBackoff = renewalRetryInitialBackoff;
+ long renewalDelay = calculateRenewalDelay(clock,
nextRenewal.get());
+ maybeScheduleRenewal(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");
}
- LOG.info(
- "Tokens update task started with {} delay",
-
TimeUtils.formatWithHighestUnit(Duration.ofMillis(renewalDelay)));
- } else {
+ } catch (InterruptedException e) {
+ // Ignore, may happen if shutting down.
+ LOG.debug("Interrupted", e);
+ } catch (Exception e) {
+ long delay = calculateRetryDelay(clock);
+ maybeScheduleRenewal(delay);
LOG.warn(
- "Tokens update task not started because either no
tokens obtained or none of the tokens specified its renewal date");
+ "Failed to update tokens, will try again in {}",
+
TimeUtils.formatWithHighestUnit(Duration.ofMillis(delay)),
+ e);
}
- } catch (InterruptedException e) {
- // 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),
- delay,
- TimeUnit.MILLISECONDS);
+ }
+ }
+
+ /**
+ * Schedules a one-shot token-obtain-and-broadcast cycle after {@code
delayMs}, replacing any
+ * pending renewal; a delay of {@code 0} brings the next cycle forward to
now. Must only be
+ * called after {@link #start(Listener)} (the scheduled and IO executors
are non-null then) and
+ * while holding {@link #tokensUpdateFutureLock}.
+ */
+ @GuardedBy("tokensUpdateFutureLock")
+ private void scheduleRenewalLocked(long delayMs) {
+ stopTokensUpdate();
+ nextScheduledAtMillis = clock.millis() + delayMs;
+ try {
+ tokensUpdateFuture =
+ scheduledExecutor.schedule(
+ () -> {
+ try {
+
ioExecutor.execute(this::startTokensUpdate);
+ } catch (RejectedExecutionException e) {
+ // IO executor is shutting down: drop the
cycle but release the
+ // dedupe flag so it cannot get stuck if
the manager is reused.
+ synchronized (tokensUpdateFutureLock) {
Review Comment:
Missed that part so this is fine.
--
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]
