Dian Fu created FLINK-41108:
-------------------------------
Summary: 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
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)
