[
https://issues.apache.org/jira/browse/FLINK-34625?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18128669#comment-18128669
]
YifanWu commented on FLINK-34625:
---------------------------------
I reproduced the same behavior with the unmodified Apache Flink 1.20.5 packages
and Apache Beam 2.48.0, and traced it to the Python read cache.
Environment: CPython 3.11.17, Java 11.0.32.1, macOS arm64, official CLI local
MiniCluster, streaming/process mode, HashMap state backend, parallelism 1. Beam
uses its pure-Python fallback in this environment.
With a 5-second TTL, OnCreateAndWrite and NeverReturnExpired, the minimal
example below writes 100, reads it immediately, waits 6 seconds and reads it
again:
* Default python.state.cache-size: immediate=100, after_ttl=100 (expected None).
* The same job with python.state.cache-size=0: immediate=100, after_ttl=None.
The cause is in SynchronousKvRuntimeState.enable_time_to_live():
{code:python}
self._remote_state_backend._state_handler._state_cache._cache._max_entries = 0
{code}
Beam's StateCache now uses _max_weight to decide whether caching is enabled;
_cache is an OrderedDict. Setting _cache._max_entries does not disable it. In
the failing 1.20.5 worker, _max_weight remains 1000 and is_cache_enabled()
remains True. The same assignment is still present on master:
https://github.com/apache/flink/blob/c1e1d359c73f8ddcfb19984a994ce9e438e14177/flink-python/pyflink/fn_execution/state_impl.py
https://github.com/apache/beam/blob/v2.48.0/sdks/python/apache_beam/runners/worker/statecache.py
Proposed fix: disable the existing StateCache in place by setting _max_weight=0
and invalidate its existing entries. Keeping the same object matters because
CachingMapStateHandler already holds a reference to it. Preserve the existing
TTL-specific write-cache policy.
I propose focused regression tests using the real Beam StateCache, covering
NeverReturnExpired, OnReadAndWrite, a policy that permits caching, and the
shared MapState cache reference. Remote responses can be controlled without
sleeping in the tests. No dependency or public API changes are needed.
FLINK-37205 fixed a related cache-capacity adaptation, but did not change this
TTL cache-disabling statement.
I would like to work on this issue. Could a committer confirm the approach and
assign it to me? I plan to submit one focused master PR with the fix and
necessary regression tests. Please also consider a 1.20 backport after review.
Minimal reproducer (the 6-second sleep is only for this manual reproducer, not
the proposed regression tests):
{code:python}
import argparse
import json
import sys
import time
from pyflink.common import Configuration, Time, Types
from pyflink.datastream import RuntimeExecutionMode, StreamExecutionEnvironment
from pyflink.datastream.functions import KeyedProcessFunction
from pyflink.datastream.state import StateTtlConfig, ValueStateDescriptor
class TtlCacheRepro(KeyedProcessFunction):
def open(self, runtime_context):
ttl = (StateTtlConfig.new_builder(Time.seconds(5))
.update_ttl_on_create_and_write()
.never_return_expired()
.build())
descriptor = ValueStateDescriptor('ttl_value', Types.INT())
descriptor.enable_time_to_live(ttl)
self.state = runtime_context.get_state(descriptor)
def process_element(self, value, ctx):
self.state.update(100)
immediate = self.state.value()
time.sleep(6)
expired = self.state.value()
yield json.dumps({'immediate': immediate, 'after_ttl': expired})
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--cache-size', type=int, default=None)
args = parser.parse_args()
config = Configuration()
config.set_string('python.execution-mode', 'process')
config.set_string('python.executable', sys.executable)
config.set_string('python.client.executable', sys.executable)
config.set_string('state.backend.type', 'hashmap')
if args.cache_size is not None:
config.set_integer('python.state.cache-size', args.cache_size)
env = StreamExecutionEnvironment.get_execution_environment(config)
env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
env.set_parallelism(1)
(env.from_collection([1], type_info=Types.INT())
.key_by(lambda value: 0, key_type=Types.INT())
.process(TtlCacheRepro(), output_type=Types.STRING())
.print())
env.execute('process-state-ttl-cache-repro')
{code}
Run it with the official CLI and the same Python interpreter for client and
worker:
{code:bash}
bin/flink run -t local -pyclientexec /path/to/python3.11 -pyexec
/path/to/python3.11 -py ttl_cache_repro.py
bin/flink run -t local -pyclientexec /path/to/python3.11 -pyexec
/path/to/python3.11 -py ttl_cache_repro.py --cache-size 0
{code}
> TTL doesn't seem to work in pyflink
> ------------------------------------
>
> Key: FLINK-34625
> URL: https://issues.apache.org/jira/browse/FLINK-34625
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Affects Versions: 1.18.1
> Environment: Image used: flink:1.18.1-scala_2.12-java11
> Reporter: Mark Lidenberg
> Priority: Major
>
> I've made a simple example to test the ttl and couldn't get the expected
> results. I went further and replicated this example in Java and it worked
> just fine. There is an inconsistency in behavior, so there is something wrong
> in pyflink or my pyflink setup.
> Here is a code to reproduce. In the example I create a state with ttl 1
> second and then process events every 1.5 seconds and print current state. I
> expect it to print `None, None, None, ...` (because ttl expires after 1.5
> seconds), but instead it prints `None, "state", "state, ...`. In Java it
> works as expected, prints `Null, Null, ...`
> ```python
> import time
> from pyflink.common import Time, Types
> from pyflink.datastream import KeyedProcessFunction, RuntimeContext,
> StreamExecutionEnvironment
> from pyflink.datastream.state import StateTtlConfig, ValueStateDescriptor
> class Processor(KeyedProcessFunction):
> def open(self, runtime_context: RuntimeContext):
> state_descriptor = ValueStateDescriptor(
> name="my_state",
> value_type_info=Types.STRING(),
> )
> state_descriptor.enable_time_to_live(
> ttl_config=StateTtlConfig.new_builder(Time.seconds(1))
> .set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite)
>
> .set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
> .build()
> )
> self.state = runtime_context.get_state(state_descriptor)
> def process_element(self, value: int, ctx: KeyedProcessFunction.Context):
> # Print current state
> print(self.state.value())
> # expect to print `None` all the time, but prints: `None, 'state',
> 'state', ...` instead
> # Update state
> self.state.update("state")
> # sleep to reset the state
> time.sleep(1.5)
> if __name__ == "__main__":
> # Init environment
> environment =
> StreamExecutionEnvironment.get_execution_environment().set_parallelism(1)
> # Setup pipeline
> (
> environment.from_collection(
> collection=list(range(10)),
> )
> .key_by(lambda value: 0)
> .process(Processor())
> )
> # Execute pipeline
> environment.execute("ttl_test")
> ```
>
> ```java
> import org.apache.flink.api.common.state.StateTtlConfig;
> import org.apache.flink.api.common.state.ValueState;
> import org.apache.flink.api.common.state.ValueStateDescriptor;
> import org.apache.flink.api.common.time.Time;
> import org.apache.flink.configuration.Configuration;
> import org.apache.flink.metrics.Histogram;
> import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
> import org.apache.flink.util.Collector;
> import java.io.IOException;
> import java.time.LocalDateTime;
> public class Processor extends KeyedProcessFunction<Integer, String, String> {
> private transient ValueState<String> state;
> @Override
> public void open(Configuration parameters) {
> var stateTtlConfig = StateTtlConfig
> .newBuilder(Time.seconds(1))
> .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
>
> .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
> .build();
> var stateDescriptor = new ValueStateDescriptor<>("state",
> String.class);
> stateDescriptor.enableTimeToLive(stateTtlConfig);
> state = getRuntimeContext().getState(stateDescriptor);
> }
> @Override
> public void processElement(String event, Context context,
> Collector<String> collector) throws IOException, InterruptedException {
> // print state
> var state = state.value();
> System.out.println(state); # prints `Null, Null, ...`
> // update state
> state.update(LocalDateTime.now().toString());
> // sleep to reset the state
> Thread.sleep(1500);
> }
> }
> ```
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
