w-yfan opened a new pull request, #29576: URL: https://github.com/apache/flink/pull/29576
## What is the purpose of the change Fix [FLINK-34625](https://issues.apache.org/jira/browse/FLINK-34625). PyFlink attempts to disable Beam's read cache through an obsolete `_cache._max_entries` attribute. Reads can therefore return cached state without reaching Flink's TTL checks or refreshing timestamps for `OnReadAndWrite`. Following Dian Fu's suggestion, bypass the read cache only for states whose TTL policy requires it. Keep the shared cache available to other states. ## Brief change log - Route the affected bag-backed states and MapState through separate handlers backed by `StateCache(0)`, using the original underlying state handler. - Preserve the existing write-cache policy and include uncached MapState handlers in iterator cleanup. - Add deterministic regression tests using real Beam caches and controlled remote responses, without sleeps. ## Verifying this change - Python 3.11 with Beam 2.69.0 and 2.75.0: the 7 new regression tests fail against the unpatched implementation and pass with the fix. After rebasing, the new tests passed again on Beam 2.69.0, and the combined 15-test suite passed on Beam 2.75.0. - `python -m flake8` passes for `flink-python`. - Three existing process-mode state integration tests pass. A local CLI job using the default cache confirms that expired ValueState and MapState entries return `None`, while a non-TTL state retains its value. - Maven verification covered all 173 reactor modules across resumed runs: 172 modules completed verification in the initial and resumed runs. After a fork exit in `flink-tests`, targeted `verify` passed all 16 `SavepointFormatITCase` cases with a 4 GiB test heap; the previously completed unit and integration-test results were preserved. Local npm configuration was corrected before resuming. This was not a single successful `clean verify` invocation. Runtime checks used an existing master distribution reporting commit `533e5fc`. Its patched `state_impl.py` is byte-for-byte identical to this commit. Focused Python checks and the full Maven verification matrix were completed after rebasing onto `8d01380f`. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): yes; the affected TTL states must reach the backend on reads, while other states retain their cache. - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable - Please also assess a separate 1.20 backport. - Suggested release note: Fix PyFlink process-mode state TTL to bypass the read cache when expiration checks or read-triggered timestamp updates are required. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
