gaborgsomogyi commented on code in PR #28639:
URL: https://github.com/apache/flink/pull/28639#discussion_r4537042467
##########
flink-runtime/src/main/java/org/apache/flink/runtime/entrypoint/ClusterEntrypoint.java:
##########
@@ -498,6 +498,16 @@ protected CompletableFuture<Void>
stopClusterServices(boolean cleanupHaData) {
final Collection<CompletableFuture<Void>> terminationFutures = new
ArrayList<>(3);
+ if (delegationTokenManager != null) {
Review Comment:
What is the idea about ordering? I mean there is a clean dependency from
`BlobServer` so closing it before that can end-up in edge cases. I think this
will happen super rarely but I would stick to proper ordering so I would close
it right before `ioExecutor`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/minicluster/MiniCluster.java:
##########
@@ -1372,6 +1372,17 @@ private void terminateMiniClusterServices(boolean
cleanupHaData) throws Exceptio
Exception exception = null;
synchronized (lock) {
+ if (delegationTokenManager != null) {
Review Comment:
Same here
##########
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;
Review Comment:
ok, I'll resolve that and will check the new approach.
##########
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:
Now we have `unregisterJob` and `unregisterJobInternal` which does the same.
I'm curious is there a reason to have them both or just a leftover?
##########
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:
Thanks, that's a fair summary. Not sure how you guys plan to manage the
custom providers so we can discuss. What I'm saying is that if a provider
throws `LinkageError` then it's a permanent issue which not going to ever
change. If that's so then the solution is to fix it and deploy a new one. Token
providers are platform specific things and not managed by users so escalation
going to come no matter what. Removing `LinkageError` and fail fatal could
potentially catch such bug early before this would go to PROD. Killing all jobs
on the shared cluster is not elegant but I think even the first registration
going to fail, right? If you handle/see things differently then please share
--
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]
