lazykitty created FLINK-40812:
---------------------------------
Summary: [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.20.5, 1.19.3, 1.18.1, 1.17.2
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
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)
