Conversation
ed1172e to
84abc1d
Compare
|
The PR is currently a draft, but I'm finalizing it. It's mostly ready, just notifying you as discussed. I'll finish the voting soon and move it to 'Ready for Review' next week. |
gaborgsomogyi
left a comment
gaborgsomogyi
left a comment
There was a problem hiding this comment.
Thanks for the contribution! Even if it's a draft I wanted to add my thoughts early.
| * <p>Backs {@link | ||
| * org.apache.flink.core.security.token.DelegationTokenManagerCallback#reobtainDelegationTokens()}. | ||
| */ | ||
| default void reobtainDelegationTokens() {} |
There was a problem hiding this comment.
I think we can have such new functions unimplemented and say that job aware approach is a mush have for all managers.
There was a problem hiding this comment.
Do you suggest to drop default bodies and make all 3 (registerJob, unregisterJob, reobtainDelegationTokens) are abstract on the interface?
CONS:
- Downstream forks with custom managers break at compile.
- Will it create a precedent to going abstract instead of default which will repeat ^ (breaking compilation) in the future again?
- Asymmetry with the provider SPI.
There was a problem hiding this comment.
I've not seen any such hardcode user who has done that but we can keep this to be on the safe side 🙂
| this.listener = checkNotNull(listener, "Listener must not be null"); | ||
| synchronized (tokensUpdateFutureLock) { | ||
| checkState(tokensUpdateFuture == null, "Manager is already started"); | ||
| stopped = false; |
There was a problem hiding this comment.
I think setting this flag freewillingly is simply wrong. AFIAU the whole point here is to go through the start/stop sequence to keep consistency inside the whole manager ecosystem. Here are my thouhgts:
- Maybe we can call it
started - Set
started=trueat the end ofstartand early return if already started - Set
started=faseat the end ofstopand early return if not started
If you have something fundamentally different in mind then feel free to challange it.
There was a problem hiding this comment.
Agreed on the rename and the early returns, implemented in 69438d6. However, I used the name running because it is used more often in the codebase.
And another part I did differently: I set running = true before the inline first cycle in start(), exactly where stopped = false sat, not at the end of the method (if I understand your suggestion correctly). The first obtain runs inline and checks the flag twice, at the top of startTokensUpdate() and again in maybeScheduleRenewal() when it schedules the periodic renewal. With the flag flipped only at the end of start(), the inline cycle sees not-running, skips the whole first obtain, and never schedules the renewal. startShouldBeIdempotent and startTokensUpdateShouldScheduleRenewal lock that ordering in, both fail if the flag moves after the cycle.
Same reason inverted on the stop() side: running = false flips before the per-job unregistration below it, so a re-obtain racing shutdown cannot schedule a cycle for a session that is shutting down. That cleanup also runs unconditionally (no early return) because ResourceManager calls stop() on the failed-startup path even when start() never ran.
| lastReobtainAtMillis = NO_PREVIOUS_REOBTAIN; | ||
| } | ||
|
|
||
| for (DelegationTokenProvider provider : delegationTokenProviders.values()) { |
There was a problem hiding this comment.
The manager is already stopped and start can be called from the other side from another thread. I think that providers will stuck in unkown state under the following circumstances:
- All providers started
- Stop called but providers not yet stopped
- Start called from another thread and start some/all of them
- Stop continues and stop some/all of them
- Realize that even if functions called in proper order some of the providers are stopped
I think we must support all situations like this because the manager is the spine of authentication.
There was a problem hiding this comment.
Start called from another thread and start some/all of them
Providers have no start/restart hooks: init() runs once in the manager constructor, and start() only drives the obtain logic, so there is nothing start() could call to re-start a provider mid stop. An obtain cycle overlapping stop() is possible, but that overlap is already covered by the stop() javadoc.
But another issue is possible. The manager instance is reused across leadership sessions, so it is possible to have stop() then start() on the same manager with the same provider instances and no re-init. A provider that closes its resources in stop() (as the javadoc tells it to) is broken in the next leadership term. The SPI javadoc says stop() is "called once during manager shutdown" and the wiring does not honor that.
I see 2 options (taking into account the existing architecture/design):
1.Fan out provider.stop() only at process shutdown and let manager stop() just stop scheduling. Providers keep a single init-to-stop lifecycle, matching their single init.
Or
2.Keep per-session stop() but redefine the contract: stop() may be followed by another start(), so providers must release only per-run resources and re-acquire lazy. Smaller diff, but it changes the agreed "called once" approach and pushes restart-stuff into every provider implementation.
I implemented option 1 in 3e53c90. DelegationTokenManager.close() is the terminal teardown: it ends the session via stop(), then stops the providers exactly once, and a closed manager rejects a later start(). It is called by the component that created the manager: ClusterEntrypoint, MiniCluster, and YarnClusterDescriptor. The per-session stop() keeps the providers usable for the next leadership session. It is a single commit, so easy to rework if you prefer option 2.
WDYT?
There was a problem hiding this 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.
There was a problem hiding this comment.
- Renamed stop() to close() in [FLINK-40019][core][runtime] Rename DelegationTokenProvider.stop() to close() . Agree it meets naming API better.
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.
yes, your understanding is correct 👍
Hi @gaborgsomogyi , thanks for the early review! Ive kept this PR as a draft because I actually had the similar questions about some items you pointed out. |
fef300b to
dd46acc
Compare
|
Hi @gaborgsomogyi . Appreciate your pre-review feedback, it was really helpful. Beyond addressing it, I found and fixed a few more gaps in follow-up commits. Moving the PR to "ready for review". I kept the branch as separate logical commits on top of the commit you already reviewed instead of squashing everything, so the increments stay reviewable. Let me know if you prefer a single squashed commit instead. |
gaborgsomogyi
left a comment
gaborgsomogyi
left a comment
There was a problem hiding this comment.
Thanks for the enhancements, I've had another round
| // way. | ||
| try { | ||
| delegationTokenManager.registerJob(jobId, jobConfiguration); | ||
| } catch (Exception | LinkageError e) { |
There was a problem hiding this 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.
There was a problem hiding this 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 8c56630 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:
- 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.
- 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).
- 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.
| * @param timeout Timeout for the future to complete | ||
| * @return Future registration response | ||
| */ | ||
| default CompletableFuture<RegistrationResponse> registerJobMaster( |
There was a problem hiding this 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
There was a problem hiding this comment.
Removed in 7c131da and tests now pass the empty configuration explicitly
| * <p>Backs {@link | ||
| * org.apache.flink.core.security.token.DelegationTokenManagerCallback#reobtainDelegationTokens()}. | ||
| */ | ||
| default void reobtainDelegationTokens() {} |
There was a problem hiding this comment.
I've not seen any such hardcode user who has done that but we can keep this to be on the safe side 🙂
| 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"); |
There was a problem hiding this 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?
| // caught so a plugin classpath failure is reported the same | ||
| // way. | ||
| try { | ||
| delegationTokenManager.registerJob(jobId, jobConfiguration); |
There was a problem hiding this 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.
| } | ||
| }, | ||
| delayMs, | ||
| TimeUnit.MILLISECONDS); |
There was a problem hiding this 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?
| lastReobtainAtMillis = NO_PREVIOUS_REOBTAIN; | ||
| } | ||
|
|
||
| for (DelegationTokenProvider provider : delegationTokenProviders.values()) { |
There was a problem hiding this 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.
| } | ||
| } | ||
|
|
||
| @VisibleForTesting |
There was a problem hiding this 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.
| * 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(); |
There was a problem hiding this 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.
c226b4a to
f0d4a1b
Compare
Rename the stopped flag to running so a never-started manager rejects work like a stopped one, and make start() idempotent. A retry no longer delays a pending re-obtain cycle, a scheduler failure no longer blocks future re-obtains, and a negative renewal delay runs immediately.
…re in job registration and cleanup A LinkageError from provider plugin code (realistically a NoClassDefFoundError, the same failure class loadProviders already special-cases at init) previously slipped past the Exception-only catches on both job hooks. The ResourceManager now also wraps the rethrown registration failure into a FlinkException naming the job before it is sent back in the RegistrationResponse, so the JobMaster log points at the job and the delegation token manager instead of a bare provider exception.
…ss re-registrations and sessions A failed re-registration of an already-registered job no longer rolls the job back, so a transient provider failure during a JobMaster reconnect cannot wipe a running job's token state. A manager-held registry tracks jobs whose provider state may exist: failed unregistrations stay tracked and are retried by the stop() drain, and stop() releases the listener and all job registrations so nothing leaks across leadership sessions. start() resets the retry backoff, the re-obtain cooldown is anchored to cycle executions (as the config option documents), an in-flight obtain cycle re-checks the running state before broadcasting, and providers receive a defensive copy of the job configuration.
…process shutdown DelegationTokenManager.close() is the terminal teardown: it ends any active session via stop() and then stops all providers, exactly once, and a closed manager rejects a later start(). It is called by the component that created the manager: ClusterEntrypoint, MiniCluster, and YarnClusterDescriptor's one-shot client-side obtain. stop() stays session-scoped (cancel scheduling, release the listener, drain job registrations) and keeps the providers usable for the next ResourceManager leadership session, so the provider stop() javadoc — called at most once, at process shutdown, never on leadership changes — holds as written.
…r instead of a mutable test setter
… start/close race and cooldown clock
…ock and tokensUpdateFutureLock to schedulingLock
…ds as guarded by renewalCycleLock
…al with stop() to match its start() call
…egistrations Track jobs only after all providers accept registration, and drop them after unregistration even if cleanup fails. Document the cleanup policy and cover both failure paths in tests. Generated-by: Fable 5.1 Generated-by: Codex (GPT-6)
Require the job configuration when registering a JobMaster with the ResourceManager. Update the existing tests to pass an empty configuration.
…r token failure A failed delegation token registration must return an RPC failure and leave no usable JobMaster registration behind. Verify that retrying the same registration request succeeds and resource declarations are then accepted. Generated-by: Codex (GPT-6)
…ter token failure Run a real JobMaster against a ResourceManager whose token manager fails the first registration. Verify automatic recovery preserves the job ID and configuration, keeps job leader monitoring active without restarting it, and allows resource declarations after registration succeeds. Generated-by: Codex (GPT-6)
f0d4a1b to
8b25efa
Compare
3 participants
What is the purpose of the change
Flink's delegation token framework is cluster-scoped: a DelegationTokenProvider obtains one set of tokens for the whole cluster and has no notion of an individual job. This breaks multi-tenant setups where jobs running on a shared cluster (e.g. a session cluster) need to authenticate as different identities against the same external service. This PR implements FLIP-588: it adds per-job awareness to the provider SPI plus the runtime wiring to invoke it. All new SPI methods are default, so existing providers keep working unchanged.
Brief change log
@Experimentalinterface DelegationTokenManagerCallback with a single reobtainDelegationTokens(), handed to providers at init time so they can request an on-demand obtain-and-broadcast cycle (e.g. right after a new job registers)@Internal) gains the corresponding methods as no-op defaults, so NoOpDelegationTokenManager and other implementations are unaffectedVerifying this change
This change added tests and can be verified as follows:
Does this pull request potentially affect one of the following parts:
@Public(Evolving): SecurityOptions (@PublicEvolving) gains the newsecurity.delegation.tokens.reobtain.cooldownoption, the extended/new SPI types (DelegationTokenProvider, DelegationTokenManagerCallback) are @experimentalDocumentation
Was generative AI tooling used to co-author this PR?