[
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.
*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 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 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.
> 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.
> *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 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)
