[
https://issues.apache.org/jira/browse/FLINK-40812?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Roman Khachatryan reassigned FLINK-40812:
-----------------------------------------
Assignee: lazykitty
> [Backport][1.20] Backport FLINK-38558 to release-1.20
> -----------------------------------------------------
>
> Key: FLINK-40812
> URL: https://issues.apache.org/jira/browse/FLINK-40812
> Project: Flink
> Issue Type: Bug
> Components: Runtime / State Backends
> Affects Versions: 1.17.2, 1.18.1, 1.19.3, 1.20.5
> Environment: Flink: 1.20.5 / release-1.20
> Java: 8
> Deployment: Standalone session cluster, parallelism 1
> State backend: RocksDB
> Timer service factory: HEAP
> Incremental checkpoints: disabled
> Runtime mode: Streaming
> Window: Event-time CUMULATE (5-second step, 20-second maximum size)
> Source/Sink: DataGen / Print
> OS: Originally observed on Linux and reproduced locally on macOS.
> Reporter: lazykitty
> Assignee: lazykitty
> Priority: Major
> Fix For: 1.20.6
>
>
> Flink 1.20 is affected by the issue fixed in FLINK-38558.
> When using the RocksDB state backend with heap timers,
> RocksDBKeyedStateBackend.isSafeToReuseKVState() returns true.
> Consequently, RecordsWindowBuffer sets requiresCopy to false and may
> reuse mutable RowData keys while flushing buffered window records.
> InternalTimerServiceImpl registers a timer using the current key
> reference without copying it. When the timer priority queue is stored
> on the JVM heap, the queue retains that mutable key reference.
> Subsequent buffer iteration may overwrite the referenced key.
> When the timer fires, it can therefore restore an incorrect distinct
> bucket key and access the wrong or missing accumulator in RocksDB
> keyed state. This can produce incorrect and non-monotonic results for
> CUMULATE windows with COUNT(DISTINCT...) when distinct aggregation
> splitting is enabled.
> FLINK-38558 fixes this by requiring key copies when RocksDB is used
> with a heap-based timer priority queue.
> Reproduction conditions:
> - RocksDB state backend
> - state.backend.rocksdb.timer-service.factory = HEAP
> - table.optimizer.distinct-agg.split.enabled = true
> - CUMULATE window
> - COUNT(DISTINCT...)
> Minimal reproducer:
> SET 'execution.runtime-mode' = 'streaming';
> SET 'parallelism.default' = '1';
> SET 'state.backend' = 'rocksdb';
> SET 'state.backend.incremental' = 'false';
> SET 'state.backend.rocksdb.timer-service.factory' = 'HEAP';
> SET 'table.optimizer.agg-phase-strategy' = 'ONE_PHASE';
> SET 'table.optimizer.distinct-agg.split.enabled' = 'true';
> CREATE TABLE source_gen (
> id BIGINT,
> card_no_id AS MOD(id, 1000),
> ts AS CURRENT_TIMESTAMP,
> WATERMARK FOR ts AS ts - INTERVAL '1' SECOND
> ) WITH (
> 'connector' = 'datagen',
> 'rows-per-second' = '70',
> 'fields.id.kind' = 'sequence',
> 'fields.id.start' = '1',
> 'fields.id.end' = '60000'
> );
> CREATE TABLE print_sink (
> window_start TIMESTAMP(3),
> window_end TIMESTAMP(3),
> cnt_distinct BIGINT,
> fire_time STRING
> ) WITH (
> 'connector' = 'print'
> );
> INSERT INTO print_sink
> SELECT
> window_start,
> window_end,
> COUNT(DISTINCT card_no_id) AS cnt_distinct,
> SUBSTR(CAST(CURRENT_TIMESTAMP AS STRING), 1, 23) AS fire_time
> FROM TABLE(
> CUMULATE(
> TABLE source_gen,
> DESCRIPTOR(ts),
> INTERVAL '5' SECOND,
> INTERVAL '20' SECOND)
> )
> GROUP BY window_start, window_end;
> Observed result:
> For the same window_start, the distinct count may decrease:
> 0 -> 0 -> 350 -> 0
> The issue was reproduced repeatedly across multiple cumulative windows.
> Expected result:
> For a complete 20-second cumulative window, the count should be
> monotonically non-decreasing:
> 350 -> 700 -> 1000 -> 1000
> Control experiments:
> 1. HEAP timer + distinct split disabled:
> 350 -> 700 -> 1000 -> 1000
> 2. ROCKSDB timer + distinct split enabled:
> 350 -> 700 -> 1000 -> 1000
> 3. Flink 2.3.0 with HEAP timer + distinct split enabled:
> 350 -> 700 -> 1000 -> 1000
> Workarounds:
> - Set state.backend.rocksdb.timer-service.factory to ROCKSDB; or
> - Disable table.optimizer.distinct-agg.split.enabled.
> The affected implementation is also present in older release lines.
> This ticket specifically tracks backporting FLINK-38558 to the
> release-1.20 LTS branch, including a regression test.
> Original issue:
> FLINK-38558
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
