[
https://issues.apache.org/jira/browse/FLINK-41435?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
LuChengxiang updated FLINK-41435:
---------------------------------
Description:
*Background*
Flink periodically obtains HDFS delegation tokens and distributes them through
HadoopDelegationTokenReceiver. The receiver currently adds new credentials to
UserGroupInformation.getCurrentUser() on the thread that performs the refresh.
Long-lived Hadoop clients, especially checkpoint storage and
FileSystemJobResultStore, capture a UGI/Subject when their DFSClient is created
and use that captured identity for later RPC connections.
If the refresh path resolves to a different current UGI from the one captured
by a long-lived client, the client can continue using the startup token even
though a new token was obtained successfully. The failure appears at the 24h
HDFS token deadline as SecretManager$InvalidToken.
*Observed failure*
This was observed in a Kubernetes application-mode job using ZooKeeper-based
high availability and HDFS-backed checkpoint storage. The primary failure was
in the checkpoint storage path:
CheckpointCoordinator.initializeCheckpointLocation
-> FsCheckpointStorageAccess.initializeLocationForCheckpoint
-> HadoopFileSystem.mkdirs
-> DFSClient.mkdirs
The checkpoint storage client reported InvalidToken for the startup token after
the refresh had created a new HDFS token and the receiver reported success. The
same JVM also showed a FileSystemJobResultStore stack using the expired token
during cleanup. The job-result store is therefore a secondary long-lived client
with the same failure mode, not the primary trigger.
Not every Kubernetes application-mode job is affected. The failure is
timing-dependent: it requires a long-lived Hadoop client to have captured one
Subject/UGI lifecycle while the refresh path later resolves to a different one.
Installation idempotency is not enough to close this issue. In the observed
Kubernetes application-mode path, the entrypoint can install the security
context twice: once before fetching the packaged program, and again in
ClusterEntrypoint.startCluster(). We tested a diagnostic build in which
SecurityUtils.install skipped the second installation when it detected the same
security configuration. Startup logged `Security configuration is already
installed, skipping re-installation.` and the login user was not replaced by a
second installation. The job still failed at the 24h HDFS token deadline.
Therefore the failure is not explained solely by a second installation
replacing the login UGI.
The second installSecurityContext(...) call does not clear Hadoop's FileSystem
cache. It is preceded by Flink's FileSystem.initialize(...), which clears only
Flink's own scheme/authority cache. It does not clear Hadoop's UGI-keyed
FileSystem.CACHE or close and rebind existing FileSystem/DFSClient instances.
Clearing Flink's cache only affects later FileSystem.get(...) calls; it does not
cause checkpoint storage or the job-result store to re-bind an already-created
FileSystem/DFSClient.
*New finding: how a fresh Subject reaches a pooled Pekko worker*
A controlled Kubernetes application-mode probe instrumented the token provider,
the HDFS RPC path, and Pekko's dispatcher thread factory. The observed chain
was:
HadoopFSDelegationTokenProvider.obtainDelegationTokens()
-> freshUGI.doAs(...)
-> fs.addDelegationTokens(...)
-> DelegationTokenIssuer.addDelegationTokens(...)
-> DistributedFileSystem.getDelegationToken()
or DistributedFileSystem.getServerDefaults()
-> DFSClient.getDelegationToken()/getServerDefaults()
-> org.apache.hadoop.ipc.Client.getRpcResponse()
-> CompletableFuture.get()
-> CompletableFuture.waitingGet()
-> ForkJoinPool.managedBlock()
-> ForkJoinPool.compensatedBlock()
-> ForkJoinPool.tryCompensate()
-> ForkJoinPool.createWorker()
-> Pekko MonitorableThreadFactory.newThread()
This happened while a Pekko default-dispatcher worker was executing
freshUGI.doAs(...). The HDFS RPC was waiting on a CompletableFuture. Because
the caller was a ForkJoinWorkerThread, JDK ForkJoinPool entered its
managed-block
compensation path and created additional dispatcher workers while the fresh
Subject was active.
The new workers' Thread constructors copied the creator's
AccessControlContext, including the fresh Subject in its inherited security
context. The probe recorded the same Subject identity in the creator and in the
new workers. Later, a cluster-io worker that resolved
UserGroupInformation.getCurrentUser() observed that inherited fresh Subject,
while getLoginUser() and HadoopSecurityContext still referred to the login
Subject. This is the concrete mechanism by which the refresh path can observe a
different UGI from the security context or login UGI.
The tested Kubernetes deployment had `fs.hdfs.impl.disable.cache=true`. With
the Hadoop FileSystem cache disabled, FileSystem.get(...) creates a new
DistributedFileSystem/DFSClient and the blocking HDFS client-initialization path
above is reachable. With the cache enabled, the same call can reuse an
already-initialized FileSystem/DFSClient and may not enter the blocking path.
The cache setting is therefore a plausible trigger condition, not a requirement
of the receiver bug itself.
*YARN status and correction of earlier observations*
Earlier YARN comparisons used UGI object identity as evidence. That was not a
valid Subject comparison: two UGI wrappers can have different object identities
while wrapping the same Subject. In a controlled YARN application-mode probe,
refresh-time current and login Subject identities were equal. The provider did
run on a Pekko dispatcher worker, but all provider invocations reported no newly
created Pekko threads. The normal YARN application-mode startup path tested so
far did not reproduce the fresh-Subject worker creation observed on Kubernetes.
The YARN core-site used by that probe does not set
`fs.hdfs.impl.disable.cache`, whereas the Kubernetes configuration does. This
is consistent with YARN reusing an initialized FileSystem/DFSClient and not
reaching the blocking client-initialization path. It is not yet a complete
causal proof: a YARN AM-level full-stack capture has not been completed, and the
existing YARN evidence is limited to the controlled probe described above.
YARN is therefore not listed as a confirmed reproduction, but it is also not
claimed to be immune. The same managedBlock/createWorker path is available on
YARN if a token-provider call creates a new HDFS client and blocks on a
ForkJoinWorkerThread. A controlled YARN run with the cache setting changed is
still needed to confirm the platform comparison.
*Root cause*
HadoopDelegationTokenReceiver targets UserGroupInformation.getCurrentUser() of
the thread executing the refresh. That current UGI is resolved from the thread's
effective AccessControlContext and is not guaranteed to be the UGI held by
HadoopSecurityContext or the UGI captured by a long-lived checkpoint-storage or
job-result-store DFSClient.
In the reproduced Kubernetes path, the freshUGI.doAs() window caused a new
Pekko dispatcher worker to inherit the fresh Subject. Consequently, the receiver
could execute with a current UGI that differed from the security-context and
login UGIs. The second security installation is not required for this mismatch,
and install idempotency therefore does not address it.
This description corrects the earlier claim that YARN had a confirmed
receiver-target Subject mismatch. The YARN claim was based on UGI object ids;
the Subject ids were actually equal in the tested run.
*Proposed change*
This issue addresses the case where a token is obtained successfully but the
refreshed credentials do not reach long-lived Hadoop clients, primarily
checkpoint storage.
* Add an accessor for the UGI held by HadoopSecurityContext.
* In HadoopDelegationTokenReceiver, add refreshed credentials to the
security-context UGI, the current UGI, and the login UGI, deduplicated by
object identity. The security-context UGI is the primary target; the other
two preserve non-secured deployments and previous behavior.
* Add tests covering (a) a refresh thread whose current user differs from the
login user, and (b) a security-context UGI distinct from the current/login
user; verify that all applicable targets receive the credentials.
Out of scope for this issue:
* Preventing HDFS client initialization or blocking RPCs from occurring inside
freshUGI.doAs() or the provider's ForkJoinPool worker.
* Closing or rebinding FileSystem instances created under a transient fresh
Subject; this is a separate client-lifecycle cleanup.
* SecurityUtils.install idempotency.
The proposed receiver change avoids depending on the refresh thread's inherited
Subject. It does not by itself guarantee that a client already created under a
different Subject is rebound. That remains a separate lifecycle concern.
*Testing*
* A controlled Kubernetes application-mode probe recorded the full chain from
fs.addDelegationTokens() through Hadoop ipc Client.getRpcResponse(),
CompletableFuture.get(), ForkJoinPool.managedBlock(), tryCompensate(), and
Pekko ThreadFactory.newThread(). The creator and the new dispatcher workers
had the same fresh Subject identity.
* A later receiver invocation on a cluster-io worker observed that inherited
fresh Subject while the login and security-context Subjects remained the
original login Subject.
* A JVM probe confirmed that a fixed thread pool created under Subject A but
first submitted under Subject B runs subsequent tasks under Subject B.
* A controlled Kubernetes application-mode probe instrumented
ExecutorThreadFactory. The first cluster-io worker was created by
DefaultLeaderElectionService-leadershipOperationExecutor-thread-1, and the
first jobmanager-io worker by a cluster-io worker; both inherited the same
Kerberos Subject at creation time.
* Receiver tests as described above.
* The controlled YARN application-mode probe observed no new Pekko dispatcher
workers in the tested normal startup path. A YARN AM-level full-stack capture
remains to be completed.
* A deployed application-mode job should survive the 24h token deadline and
continue checkpointing without SecretManager$InvalidToken. Verify that the
checkpoint-storage DFSClient receives the new token, and verify the
job-result-store client separately because it can capture a different UGI.
*Acceptance criteria*
* A token refresh updates the UGI used by checkpoint storage and other
long-lived Hadoop clients.
* Refresh does not depend on the Subject inherited by a pooled dispatcher or
cluster-io worker.
* A long-running application-mode job survives the 24h HDFS token deadline.
was:
*Background*
Flink periodically obtains HDFS delegation tokens and distributes them through
HadoopDelegationTokenReceiver. The receiver currently adds new credentials to
UserGroupInformation.getCurrentUser() on the refresh thread. Long-lived Hadoop
clients, especially checkpoint storage and FileSystemJobResultStore, capture a
UGI when their DFSClient is created and use that UGI for later RPC connections.
If the refresh thread's current UGI is not the UGI captured by a long-lived
client, the client continues using the startup token even though a new token
was obtained successfully. The failure appears at the 24h HDFS token deadline
as SecretManager$InvalidToken.
*Observed failure*
This was observed in a Kubernetes application-mode job using ZooKeeper-based
high availability and HDFS-backed checkpoint storage. The primary failure was
in the checkpoint storage path:
CheckpointCoordinator.initializeCheckpointLocation
-> FsCheckpointStorageAccess.initializeLocationForCheckpoint
-> HadoopFileSystem.mkdirs
-> DFSClient.mkdirs
The checkpoint storage client reported InvalidToken for the startup token after
the refresh had created a new HDFS token and the receiver reported success. The
same JVM also showed a FileSystemJobResultStore stack using the expired token
during cleanup. The job-result store is therefore a secondary long-lived client
with the same failure mode, not the primary trigger.
Not every Kubernetes application-mode job is affected. The failure is
timing-dependent: it requires a long-lived Hadoop client to have captured one
UGI while the token refresh path later resolves to a different UGI. If those
UGIs are the same object, the refreshed credentials reach the client normally.
Installation idempotency is not enough to close this issue. In the observed
Kubernetes application-mode path, the entrypoint can install the security
context twice: once before fetching the packaged program, and again in
ClusterEntrypoint.startCluster(). We tested a diagnostic build in which
SecurityUtils.install skipped the second installation when it detected the same
security configuration. The startup log therefore contained
`Security configuration is already installed, skipping re-installation.` and
the login user was not replaced by a second installation. The job still failed
at the 24h HDFS token deadline.
This shows that the failure is not caused solely by a second installation
replacing the login UGI. The receiver must still target the UGI captured by
long-lived Hadoop clients, independently of how many times security
installation runs.
The second installSecurityContext(...) call does not clear Hadoop's FileSystem
cache. In the application-mode startup path it is preceded by Flink's
FileSystem.initialize(...), which clears only Flink's own scheme/authority
cache. It does not clear Hadoop's UGI-keyed FileSystem.CACHE or close/rebind
existing FileSystem/DFSClient instances. Clearing Flink's cache only affects
later FileSystem.get(...) calls; it does not cause checkpoint storage or the
job-result store to re-bind an already-created FileSystem/DFSClient.
The job-result store and checkpoint storage are created after the second
install in the normal application-mode path, but by different threads. The
job-result store is created during initializeServices();
FsCheckpointStorageAccess is created later by the CheckpointCoordinator.
Creation order alone does not establish a mismatch: each DFSClient captures the
Subject of the thread that creates it. A runtime probe of a healthy job found
the job-result store and checkpoint storage clients carrying the same UGI, so
this is not the primary explanation.
FLINK-27204 (1.19+) moved FileSystemJobResultStore I/O onto the shared
cluster-io executor. DefaultDelegationTokenManager uses the same executor for
periodic token refresh. A worker in that fixed thread pool captures its
AccessControlContext when the worker is created. The receiver runs on that
worker and uses UserGroupInformation.getCurrentUser().
HadoopSecurityContext is created by HadoopSecurityContextFactory from
UserGroupInformation.getLoginUser() during security installation. It stores
that particular UGI/Subject, and clients created through
SecurityContext.runSecured() are bound to it.
The critical distinction is between thread creation and token refresh execution:
the first cluster-io worker may inherit a login Subject and still observe a
different current UGI when the refresh path executes. The receiver target is
evaluated at refresh time through getCurrentUser(), not from the worker's
creation-time Subject alone.
*YARN behavior: renewal masks, but does not fix, the receiver problem*
YARN with ZooKeeper HA uses the same ZooKeeperLeaderElectionHaServices, the
same FileSystemJobResultStore on the shared ioExecutor, and the same
DefaultDelegationTokenManager. YARN is not using a different refresh
implementation.
A controlled YARN application-mode probe produced the following observations:
* The first cluster-io worker was created by
DefaultLeaderElectionService-leadershipOperationExecutor-thread-1, matching
the first-submitter chain previously observed on Kubernetes.
* The cluster-io and jobmanager-io workers inherited the same Kerberos Subject
at creation time.
* Nevertheless, during repeated refresh cycles the receiver's
UserGroupInformation.getCurrentUser() was a different UGI object from both
the login UGI and the HadoopSecurityContext UGI.
* A temporary probe using the proposed target set updated the security-context
and login UGIs as well as the current UGI; the refreshed credentials were
present on the security-context target.
Therefore, a single security installation and YARN startup order do not make
YARN immune to the receiver-target problem. The earlier startup-order-only
explanation is insufficient.
The absence of the 24h failure on normal YARN HDFS workloads is explained by
YARN's own delegation-token renewal. Flink's delegation-token documentation
describes YARN as the traditional renewer for tokens submitted with an
application: YARN distributes these tokens to containers and keeps the HDFS
token renewed while the application is running. That renewal can keep the
original HDFS token valid at the NameNode even if Flink's receiver wrote the
new credentials to a different UGI. An already-created HDFS client can
therefore continue to use the original token, masking the Flink receiver bug.
This is masking, not a correctness fix or a platform-level exemption:
* YARN is not guaranteed to renew every token. Its normal renewal support is
primarily for HDFS tokens and depends on the token and deployment setup.
* Tokens for other services, or deployments where YARN renewal is unavailable,
remain exposed to the same receiver-target mismatch.
* A 24h YARN soak is still needed to confirm the masking path end-to-end; the
absence of reports alone is not proof that YARN is unaffected.
Kubernetes and standalone deployments do not have YARN's renewal layer for this
HDFS token path, so the receiver target mismatch can surface directly at the
24h deadline.
*Version and code-path notes*
The 1.18 path had synchronous FileSystemJobResultStore I/O, so the
worker/Subject interaction was not triggered in the deployments observed.
1.15 did not have the unified delegation-token manager path. These version
differences explain the observed frequency; they do not prove that those
versions are immune.
FLINK-27204 (1.19+) moved FileSystemJobResultStore I/O onto the shared
cluster-io executor. DefaultDelegationTokenManager uses the same executor for
periodic token refresh. The provider and receiver are invoked on the refresh
path; the security-context UGI, login UGI, and current UGI can diverge.
*Root cause*
HadoopDelegationTokenReceiver targets getCurrentUser() of the refresh thread.
That target is not guaranteed to be the UGI held by HadoopSecurityContext or
the UGI captured by a long-lived checkpoint storage or job-result-store
DFSClient. YARN's own HDFS token renewal can mask this on normal YARN HDFS
workloads, but does not make the receiver behavior correct.
*Proposed change*
This issue addresses the case where a token is obtained successfully but the
refreshed credentials do not reach long-lived Hadoop clients, primarily
checkpoint storage.
* Add an accessor for the UGI held by HadoopSecurityContext.
* In HadoopDelegationTokenReceiver, add refreshed credentials to the
security-context UGI, the current UGI, and the login UGI, deduplicated by
object identity. The security-context UGI is the primary target; the other
two preserve non-secured deployments and previous behavior.
* Add tests covering (a) a refresh thread whose current user differs from the
login user, and (b) a security-context UGI distinct from the current/login
user; verify that all applicable targets receive the credentials.
Out of scope for this issue:
* The separate case where HadoopFSDelegationTokenProvider evaluates
delegation-token requirements against the wrong current UGI and does not
obtain a token at all.
* Closing FileSystem instances created under the short-lived fresh UGI; this is
separate lifecycle cleanup.
* SecurityUtils.install idempotency.
*Testing*
* A JVM probe confirmed that a fixed thread pool created under Subject A but
first submitted under Subject B runs subsequent tasks under Subject B.
* A controlled Kubernetes application-mode probe instrumented
ExecutorThreadFactory. The first cluster-io worker was created by
DefaultLeaderElectionService-leadershipOperationExecutor-thread-1, and the
first jobmanager-io worker by a cluster-io worker; both inherited the same
Kerberos Subject. This confirms the thread-inheritance behavior.
* A YARN application-mode probe with ZooKeeper HA used the same first-submitter
chain as Kubernetes. The refresh-time probe nevertheless observed that
getCurrentUser() was a different UGI object from the login and
HadoopSecurityContext UGIs across repeated refresh cycles.
* A temporary probe using the proposed target set observed the refreshed
credentials on the security-context/login target and the current target.
* Receiver tests as described above.
* A deployed application-mode job should survive the 24h token deadline and
continue checkpointing without SecretManager$InvalidToken. Verify that the
checkpoint storage DFSClient receives the new token, and verify the
job-result-store client separately because it can capture a different UGI.
* A 24h YARN soak should verify that YARN renewal masks the failure as
expected without regressing the normal HDFS path.
*Acceptance criteria*
* A token refresh updates the UGI used by checkpoint storage and other
long-lived Hadoop clients.
* Refresh does not depend on the Subject inherited by the shared cluster-io
worker.
* A long-running application-mode job survives the 24h HDFS token deadline.
> Hadoop delegation token refresh can target a different UGI than Hadoop clients
> ------------------------------------------------------------------------------
>
> Key: FLINK-41435
> URL: https://issues.apache.org/jira/browse/FLINK-41435
> Project: Flink
> Issue Type: Bug
> Components: Runtime / Configuration
> Affects Versions: 2.0.0, 1.19.0, 1.20.0, 2.1.0, 2.2.0, 2.3.0, 2.4.0
> Environment: Flink 1.19.0 and later (1.19.x, 1.20.x, 2.0 through
> 2.4), plus master (2.5-SNAPSHOT); Kubernetes and standalone application mode
> with Kerberos and HDFS delegation tokens. The mismatch is more likely when
> FileSystemJobResultStore and the token manager share the cluster-io executor
> (FLINK-27204).
> Reporter: LuChengxiang
> Priority: Major
>
> *Background*
> Flink periodically obtains HDFS delegation tokens and distributes them through
> HadoopDelegationTokenReceiver. The receiver currently adds new credentials to
> UserGroupInformation.getCurrentUser() on the thread that performs the refresh.
> Long-lived Hadoop clients, especially checkpoint storage and
> FileSystemJobResultStore, capture a UGI/Subject when their DFSClient is
> created
> and use that captured identity for later RPC connections.
> If the refresh path resolves to a different current UGI from the one captured
> by a long-lived client, the client can continue using the startup token even
> though a new token was obtained successfully. The failure appears at the 24h
> HDFS token deadline as SecretManager$InvalidToken.
> *Observed failure*
> This was observed in a Kubernetes application-mode job using ZooKeeper-based
> high availability and HDFS-backed checkpoint storage. The primary failure was
> in the checkpoint storage path:
> CheckpointCoordinator.initializeCheckpointLocation
> -> FsCheckpointStorageAccess.initializeLocationForCheckpoint
> -> HadoopFileSystem.mkdirs
> -> DFSClient.mkdirs
> The checkpoint storage client reported InvalidToken for the startup token
> after
> the refresh had created a new HDFS token and the receiver reported success.
> The
> same JVM also showed a FileSystemJobResultStore stack using the expired token
> during cleanup. The job-result store is therefore a secondary long-lived
> client
> with the same failure mode, not the primary trigger.
> Not every Kubernetes application-mode job is affected. The failure is
> timing-dependent: it requires a long-lived Hadoop client to have captured one
> Subject/UGI lifecycle while the refresh path later resolves to a different
> one.
> Installation idempotency is not enough to close this issue. In the observed
> Kubernetes application-mode path, the entrypoint can install the security
> context twice: once before fetching the packaged program, and again in
> ClusterEntrypoint.startCluster(). We tested a diagnostic build in which
> SecurityUtils.install skipped the second installation when it detected the
> same
> security configuration. Startup logged `Security configuration is already
> installed, skipping re-installation.` and the login user was not replaced by a
> second installation. The job still failed at the 24h HDFS token deadline.
> Therefore the failure is not explained solely by a second installation
> replacing the login UGI.
> The second installSecurityContext(...) call does not clear Hadoop's FileSystem
> cache. It is preceded by Flink's FileSystem.initialize(...), which clears only
> Flink's own scheme/authority cache. It does not clear Hadoop's UGI-keyed
> FileSystem.CACHE or close and rebind existing FileSystem/DFSClient instances.
> Clearing Flink's cache only affects later FileSystem.get(...) calls; it does
> not
> cause checkpoint storage or the job-result store to re-bind an already-created
> FileSystem/DFSClient.
> *New finding: how a fresh Subject reaches a pooled Pekko worker*
> A controlled Kubernetes application-mode probe instrumented the token
> provider,
> the HDFS RPC path, and Pekko's dispatcher thread factory. The observed chain
> was:
> HadoopFSDelegationTokenProvider.obtainDelegationTokens()
> -> freshUGI.doAs(...)
> -> fs.addDelegationTokens(...)
> -> DelegationTokenIssuer.addDelegationTokens(...)
> -> DistributedFileSystem.getDelegationToken()
> or DistributedFileSystem.getServerDefaults()
> -> DFSClient.getDelegationToken()/getServerDefaults()
> -> org.apache.hadoop.ipc.Client.getRpcResponse()
> -> CompletableFuture.get()
> -> CompletableFuture.waitingGet()
> -> ForkJoinPool.managedBlock()
> -> ForkJoinPool.compensatedBlock()
> -> ForkJoinPool.tryCompensate()
> -> ForkJoinPool.createWorker()
> -> Pekko MonitorableThreadFactory.newThread()
> This happened while a Pekko default-dispatcher worker was executing
> freshUGI.doAs(...). The HDFS RPC was waiting on a CompletableFuture. Because
> the caller was a ForkJoinWorkerThread, JDK ForkJoinPool entered its
> managed-block
> compensation path and created additional dispatcher workers while the fresh
> Subject was active.
> The new workers' Thread constructors copied the creator's
> AccessControlContext, including the fresh Subject in its inherited security
> context. The probe recorded the same Subject identity in the creator and in
> the
> new workers. Later, a cluster-io worker that resolved
> UserGroupInformation.getCurrentUser() observed that inherited fresh Subject,
> while getLoginUser() and HadoopSecurityContext still referred to the login
> Subject. This is the concrete mechanism by which the refresh path can observe
> a
> different UGI from the security context or login UGI.
> The tested Kubernetes deployment had `fs.hdfs.impl.disable.cache=true`. With
> the Hadoop FileSystem cache disabled, FileSystem.get(...) creates a new
> DistributedFileSystem/DFSClient and the blocking HDFS client-initialization
> path
> above is reachable. With the cache enabled, the same call can reuse an
> already-initialized FileSystem/DFSClient and may not enter the blocking path.
> The cache setting is therefore a plausible trigger condition, not a
> requirement
> of the receiver bug itself.
> *YARN status and correction of earlier observations*
> Earlier YARN comparisons used UGI object identity as evidence. That was not a
> valid Subject comparison: two UGI wrappers can have different object
> identities
> while wrapping the same Subject. In a controlled YARN application-mode probe,
> refresh-time current and login Subject identities were equal. The provider did
> run on a Pekko dispatcher worker, but all provider invocations reported no
> newly
> created Pekko threads. The normal YARN application-mode startup path tested so
> far did not reproduce the fresh-Subject worker creation observed on
> Kubernetes.
> The YARN core-site used by that probe does not set
> `fs.hdfs.impl.disable.cache`, whereas the Kubernetes configuration does. This
> is consistent with YARN reusing an initialized FileSystem/DFSClient and not
> reaching the blocking client-initialization path. It is not yet a complete
> causal proof: a YARN AM-level full-stack capture has not been completed, and
> the
> existing YARN evidence is limited to the controlled probe described above.
> YARN is therefore not listed as a confirmed reproduction, but it is also not
> claimed to be immune. The same managedBlock/createWorker path is available on
> YARN if a token-provider call creates a new HDFS client and blocks on a
> ForkJoinWorkerThread. A controlled YARN run with the cache setting changed is
> still needed to confirm the platform comparison.
> *Root cause*
> HadoopDelegationTokenReceiver targets UserGroupInformation.getCurrentUser() of
> the thread executing the refresh. That current UGI is resolved from the
> thread's
> effective AccessControlContext and is not guaranteed to be the UGI held by
> HadoopSecurityContext or the UGI captured by a long-lived checkpoint-storage
> or
> job-result-store DFSClient.
> In the reproduced Kubernetes path, the freshUGI.doAs() window caused a new
> Pekko dispatcher worker to inherit the fresh Subject. Consequently, the
> receiver
> could execute with a current UGI that differed from the security-context and
> login UGIs. The second security installation is not required for this
> mismatch,
> and install idempotency therefore does not address it.
> This description corrects the earlier claim that YARN had a confirmed
> receiver-target Subject mismatch. The YARN claim was based on UGI object ids;
> the Subject ids were actually equal in the tested run.
> *Proposed change*
> This issue addresses the case where a token is obtained successfully but the
> refreshed credentials do not reach long-lived Hadoop clients, primarily
> checkpoint storage.
> * Add an accessor for the UGI held by HadoopSecurityContext.
> * In HadoopDelegationTokenReceiver, add refreshed credentials to the
> security-context UGI, the current UGI, and the login UGI, deduplicated by
> object identity. The security-context UGI is the primary target; the other
> two preserve non-secured deployments and previous behavior.
> * Add tests covering (a) a refresh thread whose current user differs from the
> login user, and (b) a security-context UGI distinct from the current/login
> user; verify that all applicable targets receive the credentials.
> Out of scope for this issue:
> * Preventing HDFS client initialization or blocking RPCs from occurring inside
> freshUGI.doAs() or the provider's ForkJoinPool worker.
> * Closing or rebinding FileSystem instances created under a transient fresh
> Subject; this is a separate client-lifecycle cleanup.
> * SecurityUtils.install idempotency.
> The proposed receiver change avoids depending on the refresh thread's
> inherited
> Subject. It does not by itself guarantee that a client already created under a
> different Subject is rebound. That remains a separate lifecycle concern.
> *Testing*
> * A controlled Kubernetes application-mode probe recorded the full chain from
> fs.addDelegationTokens() through Hadoop ipc Client.getRpcResponse(),
> CompletableFuture.get(), ForkJoinPool.managedBlock(), tryCompensate(), and
> Pekko ThreadFactory.newThread(). The creator and the new dispatcher workers
> had the same fresh Subject identity.
> * A later receiver invocation on a cluster-io worker observed that inherited
> fresh Subject while the login and security-context Subjects remained the
> original login Subject.
> * A JVM probe confirmed that a fixed thread pool created under Subject A but
> first submitted under Subject B runs subsequent tasks under Subject B.
> * A controlled Kubernetes application-mode probe instrumented
> ExecutorThreadFactory. The first cluster-io worker was created by
> DefaultLeaderElectionService-leadershipOperationExecutor-thread-1, and the
> first jobmanager-io worker by a cluster-io worker; both inherited the same
> Kerberos Subject at creation time.
> * Receiver tests as described above.
> * The controlled YARN application-mode probe observed no new Pekko dispatcher
> workers in the tested normal startup path. A YARN AM-level full-stack
> capture
> remains to be completed.
> * A deployed application-mode job should survive the 24h token deadline and
> continue checkpointing without SecretManager$InvalidToken. Verify that the
> checkpoint-storage DFSClient receives the new token, and verify the
> job-result-store client separately because it can capture a different UGI.
> *Acceptance criteria*
> * A token refresh updates the UGI used by checkpoint storage and other
> long-lived Hadoop clients.
> * Refresh does not depend on the Subject inherited by a pooled dispatcher or
> cluster-io worker.
> * A long-running application-mode job survives the 24h HDFS token deadline.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
