w-yfan commented on code in PR #29576:
URL: https://github.com/apache/flink/pull/29576#discussion_r4360869042
##########
flink-python/pyflink/fn_execution/state_impl.py:
##########
@@ -1109,8 +1117,11 @@ def _create_internal_map_state(
write_cache_size = 0
else:
write_cache_size = self._map_state_write_cache_size
+ map_state_handler = self._map_state_handler
+ if cache_type !=
SynchronousKvRuntimeState.CacheType.ENABLE_READ_WRITE_CACHE:
+ map_state_handler = self._uncached_map_state_handler
Review Comment:
Thanks @dianfu. I've addressed this in 1bca176d: MapState now receives the
effective read-cache policy, refreshes cached non-empty results when read
caching is disabled, and flushes pending mutations before checking the backend
while preserving local clear/write semantics.
I've added regression coverage for expiration, repeated checks within the
same bundle, pending mutations, and cache/broadcast compatibility. [CI for this
revision](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=80433)
has passed all 13 jobs. Could you please take another look?
--
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]
