[flink] branch release-1.15 updated: [FLINK-27223][python] Fix the state access problem when python.state.cache-size is set to 0

[di...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dianfu pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.15 by this push:
     new b070a7bb424 [FLINK-27223][python] Fix the state access problem when python.state.cache-size is set to 0
b070a7bb424 is described below

commit b070a7bb424da40ce0839a1151b110701de48a2c
Author: Dian Fu <di...@apache.org>
AuthorDate: Wed Apr 13 19:00:36 2022 +0800

    [FLINK-27223][python] Fix the state access problem when python.state.cache-size is set to 0
    
    This closes #19457.
---
 flink-python/pyflink/fn_execution/state_impl.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/flink-python/pyflink/fn_execution/state_impl.py b/flink-python/pyflink/fn_execution/state_impl.py
index 590c37d465e..38668dd4e83 100644
--- a/flink-python/pyflink/fn_execution/state_impl.py
+++ b/flink-python/pyflink/fn_execution/state_impl.py
@@ -149,7 +149,8 @@ class SynchronousBagKvRuntimeState(SynchronousKvRuntimeState, ABC):
         return self._internal_state
 
     def _maybe_clear_write_cache(self):
-        if self._cache_type == SynchronousKvRuntimeState.CacheType.DISABLE_CACHE:
+        if self._cache_type == SynchronousKvRuntimeState.CacheType.DISABLE_CACHE or \
+                self._remote_state_backend._state_cache_size <= 0:
             self._internal_state.commit()
             self._internal_state._cleared = False
             self._internal_state._added_elements = []