[
https://issues.apache.org/jira/browse/FLINK-41102?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131071#comment-18131071
]
Sangkyoon Nam commented on FLINK-41102:
---------------------------------------
[~dianfu], I'd like to take this too. At {{e206faa4ca2}} I reproduced it
against the real {{RemoteKeyedStateBackend}} with only remote state I/O
modeled. With {{python.state.cache-size=0}}, {{value()}} returns {{100}} after
{{clear()}} and an A -> B -> A key switch, and ListState keeps the cleared
element the same way. Committing the old internal state in
{{set_current_key()}} when the cache size is zero fixes both. I haven't
verified MapState, ReducingState or AggregatingState with this setup yet. I'll
add the regression test for clearing state and switching keys within one
bundle. Could you assign this to me?
> 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
> Priority: Major
>
> 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)
