Savonitar commented on code in PR #28639:
URL: https://github.com/apache/flink/pull/28639#discussion_r4106593169
##########
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:
You are right, the rejection alone does not imply shutdown. Addressed in
0f60c5c619d. Thanks for a good finding.
Now, If the scheduler rejects a submission, an independent timer retries it,
so arranging the retry doesn’t depend on the rejecting scheduler. Rejection
from a live IO executor also schedules another attempt. These retries submit
work and token acquisition still runs on the IO executor.
Submission retries use the configured initial backoff, with a one-second
minimum. To avoid flooding logs, the **first** rejection is logged at `WARN`
with its stack trace. Further rejections are logged at `DEBUG`, with a `WARN`
count summary at most once per minute during continued rejection. This resets
when a token cycle starts or the manager stops. Pending requests stay
coalesced, and stop() cancels pending retries.
Added deterministic tests for recovery, repeated rejection, shutdown, stale
rejection handlers, retry delays, and log rate limiting.
--
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]
