[
https://issues.apache.org/jira/browse/FLINK-41207?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
ASF GitHub Bot updated FLINK-41207:
-----------------------------------
Labels: pull-request-available (was: )
> 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
> Priority: Major
> Labels: pull-request-available
>
> 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)
