fix

Author: JNSimba

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/offset/PostgresOffset.java

@@ -45,27 +45,29 @@ public class PostgresOffset extends Offset {
     // used by PostgresOffsetFactory
     PostgresOffset(Map<String, String> offset) {
         Map<String, String> filtered = new HashMap<>(offset);
-        // When lsn == lsn_proc, WalPositionLocator is constructed with
-        // (lastCommitStoredLsn=Y, lastEventStoredLsn=Y). In pgoutput non-streaming mode,
-        // BEGIN and DML of the first new transaction after recovery share the same
-        // data_start as the previous COMMIT LSN (Y). WalPositionLocator puts Y into
-        // lsnSeen but sets startStreamingLsn to the new COMMIT LSN (Z), causing those
-        // DML records to be silently dropped in the actual streaming phase
-        // (Y in lsnSeen, Y != Z -> filtered).
+        // When a checkpoint is taken right after a COMMIT (state-3a), all three LSN fields
+        // converge to the same value: lsn == lsn_proc == lsn_commit.
+        // Recovering from such a checkpoint constructs WalPositionLocator(C0, C0), which
+        // causes the first new transaction's DML records (data_start=C0 in pgoutput) to be
+        // silently dropped: they are added to lsnSeen during the find phase, but
+        // startStreamingLsn is set to the next COMMIT (C1), so they are filtered in the
+        // stream phase.
         //
-        // Fix: when lsn == lsn_proc, remove lsn_proc and lsn_commit so that
-        // WalPositionLocator is constructed with lastCommitStoredLsn=null, which
-        // triggers the fast path: startStreamingLsn=firstLsnReceived=Y, switch-off
-        // happens immediately and all messages pass through.
+        // Fix: when lsn == lsn_proc == lsn_commit, remove lsn_proc and lsn_commit so that
+        // WalPositionLocator is constructed with lastCommitStoredLsn=null, which triggers
+        // the fast path: startStreamingLsn=firstLsnReceived=C0, all messages pass through.
+        //
+        // The triple-equality condition is safe: mid-transaction checkpoints (state-3b) have
+        // lsn_commit pointing to the previous commit, so lsn_commit != lsn, and this branch
+        // is not taken.
         //
-        // This is fixed upstream in Debezium via DBZ-6204 (adding Operation.COMMIT to
-        // the lastProcessedMessageType check in WalPositionLocator.resumeFromLsn):
-        // https://github.com/debezium/debezium/commit/3b5740f1a836c8b438888f2458ebb1554320bac7
         // This workaround can be removed once Debezium is upgraded to a version that
-        // includes DBZ-6204.
+        // includes DBZ-6204:
+        // https://github.com/debezium/debezium/commit/3b5740f1a836c8b438888f2458ebb1554320bac7
         String lsnVal = filtered.get(SourceInfo.LSN_KEY);
         String lsnProc = filtered.get(PostgresOffsetContext.LAST_COMPLETELY_PROCESSED_LSN_KEY);
-        if (lsnVal != null && lsnVal.equals(lsnProc)) {
+        String lsnCommit = filtered.get(PostgresOffsetContext.LAST_COMMIT_LSN_KEY);
+        if (lsnVal != null && lsnVal.equals(lsnProc) && lsnVal.equals(lsnCommit)) {
             filtered.remove(PostgresOffsetContext.LAST_COMPLETELY_PROCESSED_LSN_KEY);
             filtered.remove(PostgresOffsetContext.LAST_COMMIT_LSN_KEY);
         }