This is an automated email from the ASF dual-hosted git repository. dianfu pushed a commit to branch release-2.2 in repository https://gitbox.apache.org/repos/asf/flink.git
commit ee47d538cab1b0555ef3364e6a244c36abcd4b59 Author: Sangkyoon Nam <[email protected]> AuthorDate: Mon Nov 2 15:06:47 2026 +0900 [FLINK-41102][python] Flush pending keyed state operations on key switch without state cache (#29600) --- flink-python/pyflink/fn_execution/state_impl.py | 3 ++ .../pyflink/fn_execution/tests/test_state_impl.py | 51 ++++++++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/flink-python/pyflink/fn_execution/state_impl.py b/flink-python/pyflink/fn_execution/state_impl.py index 42cd672cc7f..0fdc1c7a58c 100644 --- a/flink-python/pyflink/fn_execution/state_impl.py +++ b/flink-python/pyflink/fn_execution/state_impl.py @@ -1184,6 +1184,9 @@ class RemoteKeyedStateBackend(object): if self._state_cache_size > 0: # cache old internal state self.cache_internal_state(encoded_old_key, state_obj) + else: + # flush pending operations such as clear() before dropping the old internal state + self.commit_internal_state(state_obj._internal_state) state_obj.namespace = None state_obj._internal_state = None self._current_key = key diff --git a/flink-python/pyflink/fn_execution/tests/test_state_impl.py b/flink-python/pyflink/fn_execution/tests/test_state_impl.py index 8522b9dde05..1583bef267b 100644 --- a/flink-python/pyflink/fn_execution/tests/test_state_impl.py +++ b/flink-python/pyflink/fn_execution/tests/test_state_impl.py @@ -479,6 +479,57 @@ class InternalSynchronousMapRuntimeStateTests(unittest.TestCase): self.assertTrue(self.state.contains('y')) +class KeySwitchWithoutStateCacheTests(unittest.TestCase): + + def setUp(self): + remote = InMemoryStateHandler() + self.handler = GlobalCachingStateHandler(StateCache(0), remote) + self.backend = RemoteKeyedStateBackend( + self.handler, StrUtf8Coder(), StrUtf8Coder(), 0, 16, 16) + + def test_clear_is_flushed_before_key_switch(self): + with self.handler.process_instruction_id('bundle', []): + self.backend.set_current_key('a') + value_state = self.backend.get_value_state('value', VarIntCoder()) + list_state = self.backend.get_list_state('list', VarIntCoder()) + value_state.update(100) + list_state.add(100) + self.backend.commit() + + value_state.clear() + list_state.clear() + self.assertIsNone(value_state.value()) + self.assertEqual(list(list_state.get()), []) + + self.backend.set_current_key('b') + self.assertIsNone(value_state.value()) + self.assertEqual(list(list_state.get()), []) + + self.backend.set_current_key('a') + self.assertIsNone(value_state.value()) + self.assertEqual(list(list_state.get()), []) + + +class InMemoryStateHandler: + + def __init__(self): + self.values = {} + + @contextlib.contextmanager + def process_instruction_id(self, bundle_id): + yield + + def get_raw(self, state_key, continuation_token=None): + return self.values.get(state_key.SerializeToString(), b''), None + + def append_raw(self, state_key, data): + key = state_key.SerializeToString() + self.values[key] = self.values.get(key, b'') + data + + def clear(self, state_key): + self.values.pop(state_key.SerializeToString(), None) + + class InMemoryMapStateHandler: def __init__(self):
