This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new 43b348d0fd2 [FLINK-41102][python] Flush pending keyed state operations
on key switch without state cache (#29600)
43b348d0fd2 is described below
commit 43b348d0fd2fb9ec503a8169df3d0178f7ecf576
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)
Generated-by: Claude Code 2.1.308 (Claude Opus 5.5)
---
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):
