Bowen Li created FLINK-40135:
--------------------------------
Summary: [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
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)
