[
https://issues.apache.org/jira/browse/FLINK-41108?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131094#comment-18131094
]
Sangkyoon Nam commented on FLINK-41108:
---------------------------------------
[~dianfu], I'd like to take this one as well. At {{e206faa4ca2}}, using the
real {{RemoteKeyedStateBackend}} and Beam cache with remote state I/O
(including the namespace merge) modeled, I reproduced the stale target read
with {{OnCreateAndWrite}} and {{ReturnExpiredIfNotCleanedUp}}: after the merge
the target reads {{[]}} with TTL and {{[1]}} without it. Passing the state's
TTL config to {{get_bag_state_key()}} in {{clear_state_cache()}} makes both
read {{[1]}}. I'll add a backend-level regression test for reading the cached
target after a namespace merge, with and without TTL. Could you assign this to
me?
> PyFlink namespace merges leave stale cached values for TTL-enabled ListState
> ----------------------------------------------------------------------------
>
> Key: FLINK-41108
> URL: https://issues.apache.org/jira/browse/FLINK-41108
> Project: Flink
> Issue Type: Bug
> Components: API / Python
> Reporter: Dian Fu
> Priority: Major
>
> PyFlink can return stale ListState contents after merging session-window
> namespaces when TTL is enabled.
> This can be reproduced using public APIs with two records:
> - ('k', 0, 1): key 'k', event timestamp 0 ms, integer value 1.
> - ('k', 4, 2): key 'k', event timestamp 4 ms, integer value 2.
> With a 5 ms session gap, the second record causes the existing window [0, 5)
> to merge into a new window [0, 9).
> A custom trigger maintains a ListState of integer values. During on_merge(),
> it reads the new target window's initially empty state, calls
> merge_partitioned_state(), and reads the state again.
> The expected result of this second read is [1], containing the first record's
> integer value. The second record's integer value 2 has not been added yet
> because its on_element() callback runs after on_merge().
> With TTL enabled, the second read incorrectly returns the previously cached
> empty list. Disabling TTL makes the same job succeed.
> The TTL duration is one day, with OnCreateAndWrite and
> ReturnExpiredIfNotCleanedUp. No waiting or expiration is involved.
> Reproduction:
> Save the following as repro_merge_trigger.py and run it in a matching PyFlink
> development environment containing the affected state implementation.
> {code:python}
> import argparse
> import sys
> from pyflink.common import Configuration, Time, Types
> from pyflink.common.watermark_strategy import TimestampAssigner,
> WatermarkStrategy
> from pyflink.datastream import RuntimeExecutionMode,
> StreamExecutionEnvironment
> from pyflink.datastream.functions import WindowFunction
> from pyflink.datastream.state import ListStateDescriptor, StateTtlConfig
> from pyflink.datastream.window import EventTimeSessionWindows, Trigger,
> TriggerResult
> class EventTimestamp(TimestampAssigner):
> def extract_timestamp(self, value, record_timestamp):
> return value[1]
> class MergeCheckTrigger(Trigger):
> def __init__(self, ttl_enabled):
> self.descriptor = ListStateDescriptor('merge-check', Types.INT())
> if ttl_enabled:
> self.descriptor.enable_time_to_live(
> StateTtlConfig.new_builder(Time.days(1))
> .set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite)
> .set_state_visibility(
>
> StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp)
> .build())
> def on_element(self, element, timestamp, window, ctx):
> ctx.get_partitioned_state(self.descriptor).add(element[2])
> return TriggerResult.FIRE
> def can_merge(self):
> return True
> def on_merge(self, window, ctx):
> state = ctx.get_partitioned_state(self.descriptor)
> before = list(state.get())
> ctx.merge_partitioned_state(self.descriptor)
> after = list(state.get())
> print('MERGE_CHECK before=%r after=%r expected=[1]' % (before,
> after), flush=True)
> if after != [1]:
> raise AssertionError(
> 'Namespace merge returned stale cached state: expected [1],
> got %r; '
> 'target before merge was %r' % (after, before))
> def clear(self, window, ctx):
> ctx.get_partitioned_state(self.descriptor).clear()
> def on_event_time(self, time, window, ctx):
> return TriggerResult.CONTINUE
> def on_processing_time(self, time, window, ctx):
> return TriggerResult.CONTINUE
> class DescribeWindow(WindowFunction):
> def apply(self, key, window, inputs):
> yield 'window=[%s,%s) values=%s' % (
> window.start, window.end, sorted(item[2] for item in inputs))
> def main():
> parser = argparse.ArgumentParser()
> parser.add_argument('--no-ttl', action='store_true', help='Control:
> should succeed')
> 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_integer('python.state.cache-size', 1000)
> config.set_string('restart-strategy.type', 'none')
> config.set_string('jobmanager.bind-host', '127.0.0.1')
> config.set_string('taskmanager.bind-host', '127.0.0.1')
> env = StreamExecutionEnvironment.get_execution_environment(config)
> env.set_parallelism(1)
> env.set_runtime_mode(RuntimeExecutionMode.STREAMING)
> env.from_collection(
> [('k', 0, 1), ('k', 4, 2)],
> type_info=Types.TUPLE([Types.STRING(), Types.LONG(), Types.INT()])) \
> .assign_timestamps_and_watermarks(
>
> WatermarkStrategy.no_watermarks().with_timestamp_assigner(EventTimestamp())) \
> .key_by(lambda row: row[0], key_type=Types.STRING()) \
> .window(EventTimeSessionWindows.with_gap(Time.milliseconds(5))) \
> .trigger(MergeCheckTrigger(ttl_enabled=not args.no_ttl)) \
> .apply(DescribeWindow(), output_type=Types.STRING()) \
> .print()
> env.execute('TTL namespace merge cache reproduction')
> if __name__ == '__main__':
> main()
> {code}
> Run with TTL enabled:
> {code:bash}
> PYFLINK_CYTHON_ENABLED='' python repro_merge_trigger.py
> {code}
> Actual result: the job fails with:
> {code}
> MERGE_CHECK before=[] after=[] expected=[1]
> AssertionError: Namespace merge returned stale cached state: expected [1],
> got []; target before merge was []
> {code}
> Run the same job without TTL:
> {code:bash}
> PYFLINK_CYTHON_ENABLED='' python repro_merge_trigger.py --no-ttl
> {code}
> The control run succeeds:
> {code}
> MERGE_CHECK before=[] after=[1] expected=[1]
> window=[0,5) values=[1]
> window=[0,9) values=[1, 2]
> {code}
> Root cause:
> RemoteKeyedStateBackend.clear_state_cache() constructs cache-invalidation
> keys using ttl_config=None. The actual cached state keys include the TTL
> configuration in their serialized state descriptors. Consequently,
> invalidation misses the cached entries, including the empty target state read
> before the merge.
> Cache invalidation should use the same state descriptor, including TTL
> configuration, as the corresponding reads.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
