[
https://issues.apache.org/jira/browse/FLINK-41435?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
LuChengxiang updated FLINK-41435:
---------------------------------
Description:
h2. 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.
h2. 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:
{noformat}
CheckpointCoordinator.initializeCheckpointLocation
-> FsCheckpointStorageAccess.initializeLocationForCheckpoint
-> HadoopFileSystem.mkdirs
-> DFSClient.mkdirs
{noformat}
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.
In the tested Kubernetes application-mode path, the second
installSecurityContext(...) call 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.
h2. 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:
{noformat}
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()
{noformat}
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 probe observed this HDFS RPC path while the provider was running on a
ForkJoinPool worker. It did not establish whether the FileSystem was newly
created or reused; the client lifecycle is a separate deployment-dependent
condition. The path is not treated as a required condition for this receiver
change.
h2. 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.
This absence of new workers is not proof that YARN never enters the
managedBlock/createWorker path. A YARN AM-level full-stack capture has not been
completed, so the platform-specific trigger is not yet established. YARN is
therefore not listed as a confirmed reproduction, but it is also not claimed to
be immune. The code path is available on YARN if a token-provider call creates
a new HDFS client and blocks on a ForkJoinWorkerThread; that blocking-client
path still needs to be confirmed there.
h2. 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 may 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.
h2. 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.
h2. 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 does not rely solely 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.
h2. 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. This is a supporting inheritance check, not
evidence for the ForkJoinPool compensation path.
* 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.
h2. Acceptance criteria
* A token refresh updates the UGI used by checkpoint storage and other
long-lived Hadoop clients.
* Refresh delivers credentials to the security-context UGI regardless of which
Subject a pooled dispatcher or cluster-io worker inherited.
* 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 probe confirms that creating a new HDFS client while the provider runs on a
ForkJoinPool worker can enter this path. Whether a particular deployment reuses
an existing FileSystem/DFSClient depends on the Hadoop client lifecycle; that
is outside this receiver change and is not treated as a required condition.
*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 controlled YARN startup probe did not reproduce this path. A YARN AM-level
full-stack capture has not been completed, so the platform-specific trigger is
not yet established; the absence of new workers is not proof that YARN is
immune.
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 YARN AM-level full-stack capture is still needed to
confirm whether that blocking-client path can be entered there.
*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.
> 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
>
> h2. 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.
> h2. 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:
> {noformat}
> CheckpointCoordinator.initializeCheckpointLocation
> -> FsCheckpointStorageAccess.initializeLocationForCheckpoint
> -> HadoopFileSystem.mkdirs
> -> DFSClient.mkdirs
> {noformat}
> 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.
> In the tested Kubernetes application-mode path, the second
> installSecurityContext(...) call 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.
> h2. 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:
> {noformat}
> 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()
> {noformat}
> 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 probe observed this HDFS RPC path while the provider was running on a
> ForkJoinPool worker. It did not establish whether the FileSystem was newly
> created or reused; the client lifecycle is a separate deployment-dependent
> condition. The path is not treated as a required condition for this receiver
> change.
> h2. 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.
> This absence of new workers is not proof that YARN never enters the
> managedBlock/createWorker path. A YARN AM-level full-stack capture has not
> been completed, so the platform-specific trigger is not yet established. YARN
> is therefore not listed as a confirmed reproduction, but it is also not
> claimed to be immune. The code path is available on YARN if a token-provider
> call creates a new HDFS client and blocks on a ForkJoinWorkerThread; that
> blocking-client path still needs to be confirmed there.
> h2. 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 may 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.
> h2. 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.
> h2. 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 does not rely solely 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.
> h2. 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. This is a supporting inheritance check,
> not evidence for the ForkJoinPool compensation path.
> * 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.
> h2. Acceptance criteria
> * A token refresh updates the UGI used by checkpoint storage and other
> long-lived Hadoop clients.
> * Refresh delivers credentials to the security-context UGI regardless of
> which Subject a pooled dispatcher or cluster-io worker inherited.
> * A long-running application-mode job survives the 24h HDFS token deadline.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
