[
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 the new credentials
to UserGroupInformation.getCurrentUser() on the refresh thread. Long-lived
Hadoop clients, such as checkpoint storage and the job-result store, capture a
UGI when their DFSClient is created and use tokens from that UGI for later RPC
connections.
If the refresh thread's current UGI is not the UGI captured by the client, the
client continues using the startup token even after a new token was created
successfully. The job later fails with SecretManager$InvalidToken at the 24h
deadline.
*Observed failure*
This was observed in a Kubernetes application-mode job. The primary failure was
in checkpoint storage:
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.hasCleanJobResultEntryInternal
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.
An earlier test build that skipped the second security installation (`Security
configuration is already installed`) still hit the 24h deadline; install
idempotency alone is therefore not sufficient.
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 sequentially on
the same refresh thread. The provider's freshUGI.doAs() scope ends before the
receiver runs, so the receiver's getCurrentUser() reflects the Subject
inherited by the cluster-io worker, not the provider's short-lived UGI.
HadoopSecurityContext is created by HadoopSecurityContextFactory from
UserGroupInformation.getLoginUser() during security installation. It stores
that particular UGI/Subject, and clients created inside
SecurityContext.runSecured() capture it. The token refresh is different:
DefaultDelegationTokenManager dispatches the refresh with
scheduledExecutor.schedule(() -> ioExecutor.execute(this::startTokensUpdate),
...). A worker in that fixed thread pool captures its AccessControlContext when
the worker is created. If the first cluster-io task is submitted from a
different Subject lifecycle, later refresh executions on that worker see that
inherited Subject through UserGroupInformation.getCurrentUser(). The provider's
freshUGI.doAs() only wraps the provider call and ends before the receiver runs;
the receiver therefore uses the worker's inherited Subject, not the
context/login UGI. The shared cluster-io executor is what makes that
worker/Subject mismatch reachable in this path; checkpoint storage and the
job-result store are both long-lived clients whose DFSClient can remain pinned
to a UGI that does not receive the refreshed credentials.
The 1.18 path had synchronous FileSystemJobResultStore I/O, so this
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.
*Why this is normally not seen on YARN*
YARN with ZooKeeper HA uses the same ZooKeeperLeaderElectionHaServices, the
same FileSystemJobResultStore on the shared ioExecutor, and the same
DefaultDelegationTokenManager. The executor-sharing code path is therefore
present on YARN as well; YARN is not using a different token-refresh
implementation.
The difference is startup order, not the token code:
* Kubernetes application mode calls FileSystem.initialize(...) and
installSecurityContext(...) before getPackagedProgram() because it can fetch
the job jar and artifacts from a remote location.
ClusterEntrypoint.startCluster() then calls installSecurityContext(...) again.
* YARN application mode does not perform that early
filesystem-artifact/security setup. YarnApplicationClusterEntryPoint.main only
loads the configuration and obtains the packaged program, and the only security
installation is the one in ClusterEntrypoint.startCluster().
* On YARN, runCluster() and all cluster services, including the cluster-io
pool, are then started inside that single securityContext.runSecured(...). The
code path that creates and first submits work to the pool therefore runs under
the same login/security context as the security installation. There is no
earlier security installation whose Subject can be inherited by the worker, and
no second security installation after a prior one that can replace the process
login user after clients have captured it.
* Consequently, the receiver's getCurrentUser() on the refresh worker normally
resolves to the same UGI used by the security context and by long-lived
clients. This is why the 24h failure has not been reproduced on YARN.
This is a startup-order explanation, not a platform-level exemption. If a YARN
extension or plugin changes the login user, or if a first cluster-io task is
submitted from a different Subject lifecycle, the same receiver-target issue
could occur. YARN is therefore not listed as a confirmed affected deployment
type, but it is not claimed to be immune.
*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.
*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 and login UGI, with the current UGI as a fallback,
deduplicated by object identity.
* Add a test where the refresh thread's current user differs from the
security-context user, and verify that the security-context/login UGI receives
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 runtime probe on a Kubernetes application-mode JVM confirmed that
cluster-io threads retain the Subject present when the worker is created; the
refresh receiver runs on that inherited Subject and does not re-read the
security context for each refresh.
* Source comparison: YARN application mode has one security installation in
ClusterEntrypoint.startCluster() and no Kubernetes-style early installation
before getPackagedProgram(), so the two-Subject startup condition is absent in
the normal YARN path.
* Receiver test 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; the job-result-store
client should receive the same credentials because it uses the same target 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 the shared 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 the new credentials
to UserGroupInformation.getCurrentUser() on the refresh thread. Long-lived
Hadoop clients, such as checkpoint storage and the job-result store, capture a
UGI when their DFSClient is created and use tokens from that UGI for later RPC
connections.
If the refresh thread's current UGI is not the UGI captured by the client, the
client continues using the startup token even after a new token was created
successfully. The job later fails with SecretManager$InvalidToken at the 24h
deadline.
*Observed failure*
This was observed in a Kubernetes application-mode job. The primary failure was
in checkpoint storage:
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.hasCleanJobResultEntryInternal
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.
An earlier test build that skipped the second security installation (`Security
configuration is already installed`) still hit the 24h deadline; install
idempotency alone is therefore not sufficient.
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 sequentially on
the same refresh thread. The provider's freshUGI.doAs() scope ends before the
receiver runs, so the receiver's getCurrentUser() reflects the Subject
inherited by the cluster-io worker, not the provider's short-lived UGI.
HadoopSecurityContext is created by HadoopSecurityContextFactory from
UserGroupInformation.getLoginUser() during security installation. It stores
that particular UGI/Subject, and clients created inside
SecurityContext.runSecured() capture it. The token refresh is different:
DefaultDelegationTokenManager dispatches the refresh with
scheduledExecutor.schedule(() -> ioExecutor.execute(this::startTokensUpdate),
...). A worker in that fixed thread pool captures its AccessControlContext when
the worker is created. If the first cluster-io task is submitted from a
different Subject lifecycle, later refresh executions on that worker see that
inherited Subject through UserGroupInformation.getCurrentUser(). The provider's
freshUGI.doAs() only wraps the provider call and ends before the receiver runs;
the receiver therefore uses the worker's inherited Subject, not the
context/login UGI. The shared cluster-io executor is what makes that
worker/Subject mismatch reachable in this path; checkpoint storage and the
job-result store are both long-lived clients whose DFSClient can remain pinned
to a UGI that does not receive the refreshed credentials.
The 1.18 path had synchronous FileSystemJobResultStore I/O, so this
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.
YARN with ZooKeeper HA still uses ZooKeeperLeaderElectionHaServices and the
same ioExecutor for FileSystemJobResultStore, and checkpoint storage captures
the security-context UGI through the same path as in application mode. The
JRS/executor-sharing code path therefore exists on YARN as well. The startup
order differs: YarnApplicationClusterEntrypoint does not install the security
context before getPackagedProgram(); the only installation is in
ClusterEntrypoint.startCluster(), before runCluster() starts the cluster
services. In the normal YARN application-mode path, the first cluster-io tasks
are submitted from threads already running under the login/security context, so
the worker's inherited Subject is normally the context/login UGI. This is why a
frequent YARN failure would be surprising and why YARN is not listed as a
confirmed affected deployment type, but it is not a platform-level exemption:
the decisive condition remains the Subject inherited by the first worker.
*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.
*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 and login UGI, with the current UGI as a fallback,
deduplicated by object identity.
* Add a test where the refresh thread's current user differs from the
security-context user, and verify that the security-context/login UGI receives
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.
* Receiver test 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; the job-result-store
client should receive the same credentials because it uses the same target 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 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 the new
> credentials to UserGroupInformation.getCurrentUser() on the refresh thread.
> Long-lived Hadoop clients, such as checkpoint storage and the job-result
> store, capture a UGI when their DFSClient is created and use tokens from that
> UGI for later RPC connections.
> If the refresh thread's current UGI is not the UGI captured by the client,
> the client continues using the startup token even after a new token was
> created successfully. The job later fails with SecretManager$InvalidToken at
> the 24h deadline.
> *Observed failure*
> This was observed in a Kubernetes application-mode job. The primary failure
> was in checkpoint storage:
> 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.hasCleanJobResultEntryInternal 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.
> An earlier test build that skipped the second security installation
> (`Security configuration is already installed`) still hit the 24h deadline;
> install idempotency alone is therefore not sufficient.
> 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 sequentially on
> the same refresh thread. The provider's freshUGI.doAs() scope ends before the
> receiver runs, so the receiver's getCurrentUser() reflects the Subject
> inherited by the cluster-io worker, not the provider's short-lived UGI.
> HadoopSecurityContext is created by HadoopSecurityContextFactory from
> UserGroupInformation.getLoginUser() during security installation. It stores
> that particular UGI/Subject, and clients created inside
> SecurityContext.runSecured() capture it. The token refresh is different:
> DefaultDelegationTokenManager dispatches the refresh with
> scheduledExecutor.schedule(() -> ioExecutor.execute(this::startTokensUpdate),
> ...). A worker in that fixed thread pool captures its AccessControlContext
> when the worker is created. If the first cluster-io task is submitted from a
> different Subject lifecycle, later refresh executions on that worker see that
> inherited Subject through UserGroupInformation.getCurrentUser(). The
> provider's freshUGI.doAs() only wraps the provider call and ends before the
> receiver runs; the receiver therefore uses the worker's inherited Subject,
> not the context/login UGI. The shared cluster-io executor is what makes that
> worker/Subject mismatch reachable in this path; checkpoint storage and the
> job-result store are both long-lived clients whose DFSClient can remain
> pinned to a UGI that does not receive the refreshed credentials.
> The 1.18 path had synchronous FileSystemJobResultStore I/O, so this
> 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.
> *Why this is normally not seen on YARN*
> YARN with ZooKeeper HA uses the same ZooKeeperLeaderElectionHaServices, the
> same FileSystemJobResultStore on the shared ioExecutor, and the same
> DefaultDelegationTokenManager. The executor-sharing code path is therefore
> present on YARN as well; YARN is not using a different token-refresh
> implementation.
> The difference is startup order, not the token code:
> * Kubernetes application mode calls FileSystem.initialize(...) and
> installSecurityContext(...) before getPackagedProgram() because it can fetch
> the job jar and artifacts from a remote location.
> ClusterEntrypoint.startCluster() then calls installSecurityContext(...) again.
> * YARN application mode does not perform that early
> filesystem-artifact/security setup. YarnApplicationClusterEntryPoint.main
> only loads the configuration and obtains the packaged program, and the only
> security installation is the one in ClusterEntrypoint.startCluster().
> * On YARN, runCluster() and all cluster services, including the cluster-io
> pool, are then started inside that single securityContext.runSecured(...).
> The code path that creates and first submits work to the pool therefore runs
> under the same login/security context as the security installation. There is
> no earlier security installation whose Subject can be inherited by the
> worker, and no second security installation after a prior one that can
> replace the process login user after clients have captured it.
> * Consequently, the receiver's getCurrentUser() on the refresh worker
> normally resolves to the same UGI used by the security context and by
> long-lived clients. This is why the 24h failure has not been reproduced on
> YARN.
> This is a startup-order explanation, not a platform-level exemption. If a
> YARN extension or plugin changes the login user, or if a first cluster-io
> task is submitted from a different Subject lifecycle, the same
> receiver-target issue could occur. YARN is therefore not listed as a
> confirmed affected deployment type, but it is not claimed to be immune.
> *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.
> *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 and login UGI, with the current UGI as a fallback,
> deduplicated by object identity.
> * Add a test where the refresh thread's current user differs from the
> security-context user, and verify that the security-context/login UGI
> receives 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 runtime probe on a Kubernetes application-mode JVM confirmed that
> cluster-io threads retain the Subject present when the worker is created; the
> refresh receiver runs on that inherited Subject and does not re-read the
> security context for each refresh.
> * Source comparison: YARN application mode has one security installation in
> ClusterEntrypoint.startCluster() and no Kubernetes-style early installation
> before getPackagedProgram(), so the two-Subject startup condition is absent
> in the normal YARN path.
> * Receiver test 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; the job-result-store
> client should receive the same credentials because it uses the same target
> 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 the shared 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)
