[jira] [Commented] (FLINK-40853) [postgres] Data loss: LSN commit passes the latest completed checkpoint when the delayed checkpoint has no stream offset

["kim (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-40853?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18121038#comment-18121038 ] 

kim commented on FLINK-40853:
-----------------------------

I would like to work on this. I have a fix with the tests above ready and will open a PR shortly. Could someone please assign this issue to me?

> [postgres] Data loss: LSN commit passes the latest completed checkpoint when the delayed checkpoint has no stream offset
> ------------------------------------------------------------------------------------------------------------------------
>
>                 Key: FLINK-40853
>                 URL: https://issues.apache.org/jira/browse/FLINK-40853
>             Project: Flink
>          Issue Type: Bug
>          Components: Flink CDC
>    Affects Versions: cdc-3.2.0, cdc-3.6.0
>            Reporter: kim
>            Priority: Critical
>
> h3. Summary
> With {{scan.lsn-commit.checkpoints-num-delay}} (default 3), the Postgres incremental source commits the offset of an older checkpoint to the replication slot. If that checkpoint has no stream offset, {{PostgresStreamFetchTask#commitCurrentOffset(null)}} falls back to the {{LAST_COMMIT_LSN}} of the live Debezium offset context. That LSN can be ahead of the offset of the latest completed checkpoint. PostgreSQL never resends changes before {{confirmed_flush_lsn}}, so after restoring from that checkpoint the transactions in between are lost for good. Nothing is logged or reported.
> h3. When it happens
> A checkpoint has no stream offset in {{IncrementalSourceReaderWithCommit#lastCheckpointOffsets}} in two cases. Because of the commit delay, such a checkpoint is still committed after the stream phase has started.
> # The checkpoint was taken before the stream split was assigned, for example during the snapshot phase of the {{initial}} startup mode. The first checkpoints after switching to the stream phase (3 with the default delay) commit the live LSN.
> # The checkpoint was taken while newly added tables were being snapshotted with {{scan.newly-added-table.enabled}}. {{PostgresSourceReader#snapshotState}} removes its offset. Once offset commit is turned back on, the delayed checkpoints commit the live LSN.
> A failover in these windows loses the changes between the checkpoint offset and the committed LSN.
> h3. Cause
> * [#2539|https://github.com/apache/flink-cdc/pull/2539] (cdc-3.1.0) changed {{commitCurrentOffset}} to commit the checkpoint's LSN instead of the live one, because committing the live LSN loses data after a failover. Its comment explains exactly this scenario. It kept the live LSN only as a fallback when the offset is {{null}}.
> * FLINK-35129 (cdc-3.2.0) added the commit delay. The delay can pick a checkpoint without a stream offset, so the {{null}} fallback is actually used and brings back the data loss #2539 fixed.
> The only path that moves the slot is {{commitCurrentOffset}} -> {{PostgresStreamingChangeEventSource#commitOffset}} -> {{ReplicationStream#flushLsn}}.
> h3. Reproduction
> Tests added to {{PostgresSourceReaderTest}}, using the PostgreSQL test container (PostgreSQL 14):
> * {{testLsnCommitDoesNotPassLatestCompletedCheckpoint}}: checkpoints 1-3 are taken before the stream split is assigned, checkpoint 4 in the stream phase after change 3001, then change 3002 is read and checkpoints 1-7 complete. On master, the offset of checkpoint 4 is {{0/17B98F0}}, but the slot moves to {{0/17B9A70}}.
> * {{testNoDataLossWhenRestoringAfterLsnCommit}}: same as above, then restores from checkpoint 4 and inserts 3003. On master only 3003 arrives. 3002 is lost.
> * {{testLsnCommitDoesNotPassLatestCompletedCheckpointAfterNewlyAddedTables}}: offset commit is turned off by {{OffsetCommitEvent}} for checkpoints 4-6 and on again for checkpoint 7, then change 3002 is read and checkpoint 7 completes. On master, the offset of checkpoint 7 is {{0/17B9928}}, but the slot moves to {{0/17B9AA8}}.
> All three fail on master and pass with the fix below. We also hit the first case with a real job (initial startup, TaskManager killed right after the switch to the stream phase), which lost a transaction every time in 4 runs.
> This is a client-side issue: Flink CDC reports an LSN it has not checkpointed. It does not depend on the PostgreSQL version.
> h3. Workaround
> Set {{scan.lsn-commit.checkpoints-num-delay}} to 0.
> h3. Proposed fix
> Skip the commit in {{commitCurrentOffset}} when the checkpoint has no stream offset. The slot then only receives offsets of completed checkpoints, and catches up at the next checkpoint that has one.
> h3. Not covered
> FLINK-39265 is a different data loss after recovery, caused by {{WalPositionLocator}} filtering when reading. It is independent of this issue: with the fix, the restore test above receives 3002 through the same recovery path.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)