[
https://issues.apache.org/jira/browse/FLINK-40135?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Bowen Li reassigned FLINK-40135:
--------------------------------
Assignee: Bowen Li
> [PyFlink] Prevent Beam state access after runner shutdown
> ---------------------------------------------------------
>
> Key: FLINK-40135
> URL: https://issues.apache.org/jira/browse/FLINK-40135
> Project: Flink
> Issue Type: Bug
> Components: API / Python, Runtime / State Backends
> Affects Versions: 1.19.3
> Reporter: Bowen Li
> Assignee: Bowen Li
> Priority: Major
> Fix For: 2.4.0
>
>
> Problem
> During cancellation or shutdown of a stateful PyFlink task, Beam state
> callbacks can overlap Flink state backend teardown. A late or in-flight
> callback may access keyed/operator state after RocksDB native resources have
> already been closed.
> Root cause
> - BeamStateRequestHandler previously had no lifecycle gate. It could continue
> accepting state requests while BeamPythonFunctionRunner was closing.
> - BeamPythonFunctionRunner.close() also did not close or drain the state
> request handler before Flink disposed the keyed state backend.
> As a result, Beam state access could race with RocksDB teardown.
> Expected behavior
> * Beam request production stops before the state handler closes.
> * Existing state requests finish before handler close returns.
> * New state requests after close fail fast.
> * No Beam state callback accesses Flink state after runner shutdown completes.
> Proposed change
> * Add a lifecycle gate to BeamStateRequestHandler.
> * Use a read lock while handling state requests.
> * Use a write lock during close so shutdown waits for in-flight requests.
> * Reject new requests after the handler is closed.
> * Close JobBundleFactory first, then close/drain BeamStateRequestHandler.
> * Add tests for close draining, post-close rejection, and runner shutdown
> ordering.
> Impact
> This race can cause intermittent failures or native RocksDB crashes during
> PyFlink task cancellation/failover and prolong recovery.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
