[
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 capture a UGI when their DFSClient is created, and subsequent
RPC connections use tokens from that captured UGI.
If the refresh thread's current UGI is not the UGI captured by the client, the
client continues using the startup token even though a new token was obtained
successfully. The job later fails with SecretManager$InvalidToken when the
startup token reaches its 24h deadline.
*How it is triggered*
This was observed in a Kubernetes application-mode job with no
connector-specific security setup. A token refresh generated a new HDFS token
and the receiver logged success, but FileSystemJobResultStore and checkpoint
storage continued using the token created at startup. A build containing the
install-idempotency change still failed: startup logged `Security configuration
is already installed` and did not reinstall, but the 24h expiry remained.
FLINK-27204 (1.19+) moved FileSystemJobResultStore I/O onto the shared
cluster-io executor. DefaultDelegationTokenManager uses the same executor for
periodic token refresh. Java thread-pool workers inherit the
AccessControlContext/Subject of the thread that first creates the worker, not
the Subject at executor construction. A worker can therefore see a different
UGI through getCurrentUser() than the login/security-context UGI captured by
the Hadoop client.
This is consistent with why the older 1.18 path did not show the issue:
FileSystemJobResultStore I/O was synchronous, and the token-refresh executor
was not shared with JRS I/O. 1.15 did not have the unified delegation-token
manager path at all.
*Root cause*
HadoopDelegationTokenReceiver targets getCurrentUser() of the refresh thread.
That target is not guaranteed to be the UGI held by the HadoopSecurityContext
or the UGI captured by a long-lived FileSystemJobResultStore / checkpoint
DFSClient. HadoopFSDelegationTokenProvider also performs its
Kerberos/token-required checks using the current user and creates a short-lived
UGI for token acquisition.
*Proposed change*
* Add an accessor for the UGI held by HadoopSecurityContext.
* In HadoopDelegationTokenReceiver, add refreshed credentials to the
security-context UGI, current UGI, and login UGI, deduplicated by object
identity.
* In HadoopFSDelegationTokenProvider, resolve delegation-token necessity under
the security-context UGI and close FileSystem instances created under the
short-lived login UGI.
* Add regression coverage for a refresh thread whose current user differs from
the security-context user.
*Scope*
This issue covers receiver/provider UGI targeting and client credential
visibility. It does not cover SecurityUtils.install idempotency,
partial-install rollback, or arbitrary later changes to the process login user
by user code.
*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.
* Unit/integration coverage should verify that refreshed credentials reach the
UGI held by HadoopSecurityContext and the login/current UGIs used by long-lived
clients.
* A deployed build with the focused fix should survive the 24h token deadline
and continue checkpointing without SecretManager$InvalidToken.
*Acceptance criteria*
* Token refresh updates the UGI used by FileSystemJobResultStore and checkpoint
DFSClient.
* Refresh does not depend on the Subject inherited by the shared cluster-io
worker.
* A long-running job survives the 24h HDFS token deadline.
was:
*Background*
In Kubernetes and standalone application mode, Flink installs the security
context twice:
* KubernetesApplicationClusterEntrypoint and
StandaloneApplicationClusterEntryPoint call
ClusterEntrypoint.installSecurityContext() before fetching user artifacts.
* ClusterEntrypoint.startCluster() calls installSecurityContext() again.
Both calls are part of the normal startup path. Each call creates a new
SecurityConfiguration and runs the security modules and context factory again.
With Kerberos this performs another login and can replace the process login UGI.
Components created under the first UGI, for example Hadoop clients used while
fetching artifacts, then remain bound to a UGI that no longer receives
refreshed delegation tokens. The refresh path can update a different UGI than
the one used by those clients, so token refresh can silently stop; the job
later fails with errors such as SecretManager$InvalidToken or an expired HDFS
delegation token.
*How it is triggered*
This happens in Kubernetes and standalone application mode when Kerberos is
enabled. The first install is required before fetching user artifacts; the
second install is part of ClusterEntrypoint.startCluster(). The second call
cannot simply be removed, because startCluster() still needs a security context.
A naive comparison of the two Configuration objects would also not help:
ApplicationClusterEntryPoint.configureExecution() adds pipeline.jars,
pipeline.classpaths and execution.target between the two calls. The
security-relevant options are unchanged, so the fix must compare only the
options consumed by the security modules.
*Root cause*
SecurityUtils.install() unconditionally runs the security modules and creates a
new security context. It does not remember the last successfully installed
configuration, so an equivalent configuration is treated as a real
re-installation.
*Proposed change*
* Give SecurityConfiguration value semantics for the options consumed by the
bundled security modules: security.*, zookeeper.sasl.*, flink.hadoop.*,
fs.hdfs.* and io.tmp.dirs. Include the configured security context and module
factories.
* Remember the last successfully installed SecurityConfiguration in
SecurityUtils.
* Make SecurityUtils.install() synchronized and return early when the new
configuration equals the installed one.
* Clear the remembered configuration in SecurityUtils.uninstall().
* Do not reinstall because of unrelated runtime options such as pipeline.jars,
pipeline.classpaths or execution.target.
*Scope*
This issue only covers idempotent installation. Delegation token
receiver/provider target-UGI handling, partial-install rollback and file-system
cleanup during token acquisition are out of scope. The fix keeps a single UGI
for the two Flink install calls; it does not make the process resilient to
arbitrary later changes of the global Hadoop login user by user code or
connectors.
*Testing*
* SecurityUtilsInstallTest: an equal configuration after adding pipeline
options does not reinstall; a different security-relevant configuration does
reinstall; install after uninstall installs again.
* SecurityConfigurationTest: security-relevant prefixes affect equality;
unrelated options do not.
* Existing SecurityUtilsTest still passes.
* Verified on master (2.5-SNAPSHOT, Java 17, Hadoop 3.4.3): 16 tests pass.
*Acceptance criteria*
* Repeated install() with an equivalent security configuration keeps the
existing context and modules.
* A different security-relevant configuration still installs a new context and
modules.
* Unrelated runtime options do not trigger reinstallation.
* uninstall() clears the stored installation state.
> 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.20.0, 2.1.0, 2.2.0, 2.3.0, 2.4.0
> Environment: Flink 1.20.x and 2.0 through 2.4, plus master
> (2.5-SNAPSHOT); Kubernetes and standalone application mode with Kerberos and
> HDFS delegation tokens.
> Reporter: LuChengxiang
> Priority: Major
> Labels: pull-request-available
>
> *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 capture a UGI when their DFSClient is created, and
> subsequent RPC connections use tokens from that captured UGI.
> If the refresh thread's current UGI is not the UGI captured by the client,
> the client continues using the startup token even though a new token was
> obtained successfully. The job later fails with SecretManager$InvalidToken
> when the startup token reaches its 24h deadline.
> *How it is triggered*
> This was observed in a Kubernetes application-mode job with no
> connector-specific security setup. A token refresh generated a new HDFS token
> and the receiver logged success, but FileSystemJobResultStore and checkpoint
> storage continued using the token created at startup. A build containing the
> install-idempotency change still failed: startup logged `Security
> configuration is already installed` and did not reinstall, but the 24h expiry
> remained.
> FLINK-27204 (1.19+) moved FileSystemJobResultStore I/O onto the shared
> cluster-io executor. DefaultDelegationTokenManager uses the same executor for
> periodic token refresh. Java thread-pool workers inherit the
> AccessControlContext/Subject of the thread that first creates the worker, not
> the Subject at executor construction. A worker can therefore see a different
> UGI through getCurrentUser() than the login/security-context UGI captured by
> the Hadoop client.
> This is consistent with why the older 1.18 path did not show the issue:
> FileSystemJobResultStore I/O was synchronous, and the token-refresh executor
> was not shared with JRS I/O. 1.15 did not have the unified delegation-token
> manager path at all.
> *Root cause*
> HadoopDelegationTokenReceiver targets getCurrentUser() of the refresh thread.
> That target is not guaranteed to be the UGI held by the HadoopSecurityContext
> or the UGI captured by a long-lived FileSystemJobResultStore / checkpoint
> DFSClient. HadoopFSDelegationTokenProvider also performs its
> Kerberos/token-required checks using the current user and creates a
> short-lived UGI for token acquisition.
> *Proposed change*
> * Add an accessor for the UGI held by HadoopSecurityContext.
> * In HadoopDelegationTokenReceiver, add refreshed credentials to the
> security-context UGI, current UGI, and login UGI, deduplicated by object
> identity.
> * In HadoopFSDelegationTokenProvider, resolve delegation-token necessity
> under the security-context UGI and close FileSystem instances created under
> the short-lived login UGI.
> * Add regression coverage for a refresh thread whose current user differs
> from the security-context user.
> *Scope*
> This issue covers receiver/provider UGI targeting and client credential
> visibility. It does not cover SecurityUtils.install idempotency,
> partial-install rollback, or arbitrary later changes to the process login
> user by user code.
> *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.
> * Unit/integration coverage should verify that refreshed credentials reach
> the UGI held by HadoopSecurityContext and the login/current UGIs used by
> long-lived clients.
> * A deployed build with the focused fix should survive the 24h token deadline
> and continue checkpointing without SecretManager$InvalidToken.
> *Acceptance criteria*
> * Token refresh updates the UGI used by FileSystemJobResultStore and
> checkpoint DFSClient.
> * Refresh does not depend on the Subject inherited by the shared cluster-io
> worker.
> * A long-running job survives the 24h HDFS token deadline.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
