nzw921rx commented on code in PR #11556:
URL: https://github.com/apache/seatunnel/pull/11556#discussion_r4374264274
##########
seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/wal/PostgresWalFetchTask.java:
##########
@@ -118,6 +131,93 @@ public SourceSplitBase getSplit() {
return split;
}
+ /**
+ * A bounded WAL reader used to reconcile changes made while a snapshot
split is scanned.
+ *
+ * <p>The reader emits an END watermark after completely processing the
high watermark.
+ */
+ public static class PostgresWalSplitReadTask extends
PostgresStreamingChangeEventSource {
+
+ /**
+ * Defines the inclusive low watermark and inclusive high watermark
for this reader.
+ *
+ * <p>The Debezium stopping LSN is populated from this split.
+ */
+ private final IncrementalSplit walSplit;
+
+ /**
+ * Emits the END watermark into the same queue as snapshot and WAL
records.
+ *
+ * <p>The scan fetcher uses that watermark to finish its
reconciliation buffer.
+ */
+ private final JdbcSourceEventDispatcher<PostgresPartition> dispatcher;
+
+ /**
+ * Propagates failures captured internally by Debezium's streaming
source.
+ *
+ * <p>Debezium records producer failures instead of throwing them
directly.
+ */
+ private final PostgresErrorHandler errorHandler;
+
+ /**
+ * Creates a WAL reader that stops at the snapshot split's high
watermark.
+ *
+ * <p>The supplied replication connection belongs only to the current
snapshot reader.
+ */
+ public PostgresWalSplitReadTask(
+ PostgresConnectorConfig connectorConfig,
+ Snapshotter snapshotter,
+ PostgresConnection connection,
+ PostgresEventDispatcher<TableId> pgEventDispatcher,
+ JdbcSourceEventDispatcher<PostgresPartition> dispatcher,
+ PostgresErrorHandler errorHandler,
+ PostgresSchema schema,
+ PostgresTaskContext taskContext,
+ ReplicationConnection replicationConnection,
+ IncrementalSplit walSplit) {
+ super(
+ connectorConfig,
+ snapshotter,
+ connection,
+ pgEventDispatcher,
+ errorHandler,
+ Clock.SYSTEM,
+ schema,
+ taskContext,
+ replicationConnection);
+ this.walSplit = walSplit;
+ this.dispatcher = dispatcher;
+ this.errorHandler = errorHandler;
+ }
+
+ /**
+ * Reads from the low watermark until Debezium completely processes
the high watermark, then
+ * emits END so the snapshot fetcher can merge its buffered records.
+ */
+ @Override
+ public void execute(
+ ChangeEventSourceContext context,
+ PostgresPartition partition,
+ PostgresOffsetContext offsetContext)
+ throws InterruptedException {
+ offsetContext.setStreamingStoppingLsn(((LsnOffset)
walSplit.getStopOffset()).getLsn());
Review Comment:
[P1] Complete bounded backfill when the source database has no decodable
events
With `exactly_once=true` and `startup.mode=initial`, the source database can
remain idle throughout the snapshot while another database on the same
PostgreSQL instance commits transactions. `PostgresUtils.currentLsn()` reads
the cluster-wide `pg_current_wal_lsn()`, so those commits can make `HIGH > LOW`
and enter this backfill path. However, the source database's logical
replication slot does not decode the other database's transactions.
Passing this HIGH to `setStreamingStoppingLsn()` makes the reader wait for
`lastCompletelyProcessedLsn >= HIGH`, but that position advances only when a
message is decoded. If no source-database message arrives, `super.execute()`
does not return and END is never emitted. `IncrementalSourceScanFetcher` then
cannot deliver the buffered snapshot or complete the split, and the replication
slots can continue retaining WAL.
Please use a reliable backfill boundary protocol that can confirm completion
across WAL that produces no decodable source-database event, while preserving
transaction completeness. Add a regression test where only another database
generates WAL during the snapshot and the captured database remains idle.
Simply timing out and emitting END would risk reintroducing data loss.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
