Xiaobing Fang created FLINK-41207:
-------------------------------------
Summary: Fluss pipeline source does not checkpoint consumed
offsets across empty log intervals
Key: FLINK-41207
URL: https://issues.apache.org/jira/browse/FLINK-41207
Project: Flink
Issue Type: Bug
Components: Flink CDC
Reporter: Xiaobing Fang
h2. Description
The Fluss pipeline source loses consumption progress when the log scanner
advances across an interval without returning any records.
{\{MultiTableRecords.consumedUpToOffset()}} reports the exclusive consumed
offset for a bucket, even when its record list is empty. However,
\{{FlussSplitReader}} only queues business records, and \{{FlussRecordEmitter}}
only updates split state while processing those records.
The scanner can therefore advance while the checkpointed split offset remains
unchanged. After recovery, the source scans the same empty interval again.
h2. Steps to reproduce
# Create an Indexed Log table and append one record.
# Consume the record through the CDC reader. The split's next offset becomes
\{{1}}.
# Use ProduceLog RPC to append a valid batch with \{{recordCount=0}} whose
\{{nextLogOffset}} advances to \{{2}}.
# Let the scanner consume the batch. It returns zero records and
\{{consumedUpToOffset=2}}.
# Process the reader's fetched results, call \{{snapshotState()}}, and
serialize the saved split.
*Expected:* The saved and serialized split offset is \{{2}}. Recovery resumes
from offset \{{2}}.
*Actual:* The saved and serialized split offset remains \{{1}}. Recovery starts
from offset \{{1}} and scans the empty interval again.
h2. Proposed fix
Carry consumption progress through the reader queue using a progress-only
element:
* Queue it after all real records for the same bucket.
* Add it for empty buckets or when \{{consumedUpToOffset}} exceeds the last
record's offset + 1.
* Update split state only when the reader thread processes it.
* Do not emit business records, consume pending schema events, or increase the
input record count.
This also covers hybrid snapshot/log splits without changing the split
serialization format.
h2. Validation
A real Fluss cluster regression test reproduces the failure with the old reader
and passes with the fix. It verifies checkpoint timing, split serialization,
and reading a new record after restoration.
Focused tests pass on:
* Flink 1.20.3 / Java 11
* Flink 2.2.0 / Java 17
h2. Scope
The empty batch above is a deterministic reproduction. A separate SDK batching
defect can also produce such batches through normal append calls. Fixing that
producer-side defect is outside the scope of this issue.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
