Dian Fu created FLINK-41102:
-------------------------------
Summary: Keyed state clear operations are lost on key changes when
python.state.cache-size is zero
Key: FLINK-41102
URL: https://issues.apache.org/jira/browse/FLINK-41102
Project: Flink
Issue Type: Bug
Components: API / Python
Reporter: Dian Fu
When python.state.cache-size is set to 0, a pending clear() operation can be
lost if the current key changes before the state backend commits. Switching
back to the original key can then make the previously cleared state visible
again.
This issue is found during https://github.com/apache/flink/pull/29576.
Reproduction scenario
Configure python.state.cache-size=0 and use a keyed Python operator with
ValueState:
1. Store 100 for key A and complete a bundle so that the value is persisted.
2. In a subsequent bundle, process a record for key A and call state.clear().
3. Process a record for key B before the bundle is committed.
4. Process another record for key A and call state.value().
Steps 2–4 must execute without an intervening state commit.
Expected result
state.value() returns None because the state for key A was explicitly cleared.
Actual result
state.value() returns the previously persisted value, 100.
Reading the state immediately after clear(), before switching keys, correctly
returns None. The deletion is lost during the key change.
Root cause
SynchronousValueRuntimeState.clear() delegates to Beam's
SynchronousBagRuntimeState.clear(), which only updates local state:
self._cleared = True
self._added_elements = []
The remote deletion is deferred until commit().
However, RemoteKeyedStateBackend.set_current_key() only retains the previous
internal state object when the state cache size is greater than zero:
if self._state_cache_size > 0:
self.cache_internal_state(encoded_old_key, state_obj)
state_obj.namespace = None
state_obj._internal_state = None
With a zero-sized cache, the previous object is discarded without being
committed. Its pending clear marker is therefore lost. Switching back creates a
new internal state object, which reads the unchanged value from the remote
backend.
Ordinary bag-state writes handle this configuration through
_maybe_clear_write_cache(), which commits immediately when the cache size is
zero. clear() does not invoke that path.
Affected code is in flink-python/pyflink/fn_execution/state_impl.py.
The same lifecycle problem affects other keyed state implementations with
deferred clear operations, including ListState, ReducingState,
AggregatingState, and MapState.
Verification
Confirmed through source inspection and deterministic Python-level
reproductions using the real state implementations with modeled remote state
I/O.
A JVM integration regression test should cover clearing state and switching
keys within the same bundle with python.state.cache-size=0.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
