Savonitar commented on code in PR #28639:
URL: https://github.com/apache/flink/pull/28639#discussion_r3539746726
##########
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);
Review Comment:
You are right, the log could claim a delay that was never scheduled. Fixed
in 55b0566: maybeScheduleRenewal now returns the delay that actually took
effect and the log prints that.
--
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]
