This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new b43c38ae0de [FLINK-41102][python] Flush pending keyed state operations
on key switch without state cache (#29600)
b43c38ae0de is described below
commit b43c38ae0def378932232a6bff79446328a5ddb9
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 | 58 +++++++++++++++++++++-
2 files changed, 59 insertions(+), 2 deletions(-)
diff --git a/flink-python/pyflink/fn_execution/state_impl.py
b/flink-python/pyflink/fn_execution/state_impl.py
index 7b3694727b8..17a9f76c5e7 100644
--- a/flink-python/pyflink/fn_execution/state_impl.py
+++ b/flink-python/pyflink/fn_execution/state_impl.py
@@ -1141,6 +1141,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 30f42e73ff3..01dfe66a3e7 100644
--- a/flink-python/pyflink/fn_execution/tests/test_state_impl.py
+++ b/flink-python/pyflink/fn_execution/tests/test_state_impl.py
@@ -16,13 +16,16 @@
# limitations under the License.
################################################################################
+import contextlib
import unittest
-from apache_beam.coders import StrUtf8Coder
+from apache_beam.coders import StrUtf8Coder, VarIntCoder
from apache_beam.portability.api import beam_fn_api_pb2
+from apache_beam.runners.worker.sdk_worker import GlobalCachingStateHandler
+from apache_beam.runners.worker.statecache import StateCache
from pyflink.fn_execution.state_impl import CachingMapStateHandler, \
- InternalSynchronousMapRuntimeState
+ InternalSynchronousMapRuntimeState, RemoteKeyedStateBackend
class InternalSynchronousMapRuntimeStateTests(unittest.TestCase):
@@ -71,6 +74,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):
