[cdc] Flush at the beginning of CDC until consumedXLogPos

PeerDB-io · Nov 20, 2023 · a37bec5 · a37bec5
1 parent 2a60ff3
commit a37bec5
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/flow/connectors/postgres/cdc.go b/flow/connectors/postgres/cdc.go
@@ -182,6 +182,12 @@ func (p *PostgresCDCSource) consumeStream(
 	consumedXLogPos := pglogrepl.LSN(0)
 	if clientXLogPos > 0 {
 		consumedXLogPos = clientXLogPos - 1
+
+		err := pglogrepl.SendStandbyStatusUpdate(p.ctx, conn,
+			pglogrepl.StandbyStatusUpdate{WALWritePosition: consumedXLogPos})
+		if err != nil {
+			return fmt.Errorf("[initial-flush] SendStandbyStatusUpdate failed: %w", err)
+		}
 	}
 
 	var standByLastLogged time.Time