[
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 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.
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 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.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.
Not every Kubernetes application-mode job is affected. The failure is
timing-dependent: it occurs when a long-lived Hadoop client has captured one
UGI and the cluster-io worker that later performs the refresh has inherited a
different Subject. If the client and the refresh worker resolve to the same
UGI, the refreshed credentials reach that client and the job can continue
normally. The same is true if the affected client is recreated before the old
token expires.
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. This is consistent with the
condition above: removing the second installation removes one way to replace
the login UGI, but it does not guarantee that every already-created client and
the refresh worker share the same Subject.
The second installSecurityContext(...) call itself 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 FileSystem 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 or recreate a
FileSystem/DFSClient instance they already hold.
The job-result store and checkpoint storage are not created before the second
install in the normal application-mode path: the job-result store is created in
initializeServices() after installSecurityContext(), and
FsCheckpointStorageAccess is created by CheckpointCoordinator when the job
starts. They are created by different threads/executors: the job-result store
is created on the entrypoint thread in initializeServices(), while
FsCheckpointStorageAccess is created by the JobMaster on the jobmanager-io
executor. This timing and thread difference alone does not establish a UGI
mismatch. Each FileSystem/DFSClient captures the current Subject of the thread
that creates it, not the installation order; a runtime probe of an observed job
found the resulting job-result store and checkpoint storage clients carrying
the same UGI, so the JIRA does not claim a confirmed thread-Subject divergence.
The second installation also affects the installing thread and threads created
under that context; it does not rewrite the inherited AccessControlContext of
threads that already exist.
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 than the UGI captured by the long-lived client,
later refresh executions on that worker see that inherited Subject through
UserGroupInformation.getCurrentUser(). 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() as part of its
artifact-fetch path; the job jar and artifacts can be remote.
ClusterEntrypoint.startCluster() then calls installSecurityContext(...) again.
* YARN application mode does not perform that early
filesystem-artifact/security setup. YarnApplicationClusterEntryPoint.main
obtains the packaged program without an earlier security installation; 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
pool is created, and normal startup work is submitted, inside that single
secured context. There is no earlier security installation whose Subject can be
inherited by the worker, and no second installSecurityContext that can replace
the login UGI after the first one.
* 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 in the normal YARN
application-mode path.
This is a startup-order explanation, not a platform-level exemption. If a
custom security module 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, 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 that a worker's Subject is inherited from the
submitting thread. The cluster-io worker was created before the
job-result-store filesystem client in this healthy run, and both the job-result
store and checkpoint storage captured that same UGI. A second minimal-load
application-mode run with the same early image reproduced the same
first-submitter chain.
* 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 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.
*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 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.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
