You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by di...@apache.org on 2022/04/07 11:04:14 UTC
[flink] branch release-1.13 updated: [FLINK-27108][python] Fix the state cache clean up logic
This is an automated email from the ASF dual-hosted git repository.
dianfu pushed a commit to branch release-1.13
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.13 by this push:
new 66d896081ae [FLINK-27108][python] Fix the state cache clean up logic
66d896081ae is described below
commit 66d896081ae9c9572972ec13cef8169d57391444
Author: Dian Fu <di...@apache.org>
AuthorDate: Thu Apr 7 15:29:02 2022 +0800
[FLINK-27108][python] Fix the state cache clean up logic
This closes #19386.
---
flink-python/pyflink/fn_execution/state_impl.py | 3 +++
.../api/runners/python/beam/BeamPythonFunctionRunner.java | 12 +++++++-----
2 files changed, 10 insertions(+), 5 deletions(-)
diff --git a/flink-python/pyflink/fn_execution/state_impl.py b/flink-python/pyflink/fn_execution/state_impl.py
index 50a2115ed9b..42458dbf8b6 100644
--- a/flink-python/pyflink/fn_execution/state_impl.py
+++ b/flink-python/pyflink/fn_execution/state_impl.py
@@ -85,6 +85,9 @@ class LRUCache(object):
def __iter__(self):
return iter(self._cache.values())
+ def __contains__(self, key):
+ return key in self._cache
+
class SynchronousKvRuntimeState(InternalKvState, ABC):
"""
diff --git a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java
index 78dd53c0979..bbcbe5161c9 100644
--- a/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java
+++ b/flink-python/src/main/java/org/apache/flink/streaming/api/runners/python/beam/BeamPythonFunctionRunner.java
@@ -234,15 +234,17 @@ public abstract class BeamPythonFunctionRunner implements PythonFunctionRunner {
PortablePipelineOptions portableOptions =
PipelineOptionsFactory.as(PortablePipelineOptions.class);
- if (jobOptions.containsKey(PythonOptions.STATE_CACHE_SIZE.key())) {
+ int stateCacheSize =
+ Integer.parseInt(
+ jobOptions.getOrDefault(
+ PythonOptions.STATE_CACHE_SIZE.key(),
+ PythonOptions.STATE_CACHE_SIZE.defaultValue().toString()));
+ if (stateCacheSize > 0) {
portableOptions
.as(ExperimentalOptions.class)
.setExperiments(
Collections.singletonList(
- ExperimentalOptions.STATE_CACHE_SIZE
- + "="
- + jobOptions.get(
- PythonOptions.STATE_CACHE_SIZE.key())));
+ ExperimentalOptions.STATE_CACHE_SIZE + "=" + stateCacheSize));
}
Struct pipelineOptions = PipelineOptionsTranslation.toProto(portableOptions);