davidzollo commented on code in PR #10058:
URL: https://github.com/apache/seatunnel/pull/10058#discussion_r2523411740
##########
seatunnel-connectors-v2/connector-cdc/connector-cdc-postgres/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/source/reader/snapshot/PostgresSnapshotSplitReadTask.java:
##########
@@ -166,7 +166,7 @@ private void createDataEvents(PostgresSnapshotContext
snapshotContext, TableId t
EventDispatcher.SnapshotReceiver snapshotReceiver =
dispatcher.getSnapshotChangeEventReceiver();
log.debug("Snapshotting table {}", tableId);
- TableId newTableId = new TableId(null, tableId.schema(),
tableId.table());
+ TableId newTableId = new TableId(tableId.catalog(), tableId.schema(),
tableId.table());
Review Comment:
Good job.
Did you test this PR in old pg version? Probably it can't work well in the
old version. Maybe you can refer to the following code.
```
String catalog = tableId.catalog();
if (catalog == null || catalog.isEmpty()) {
log.warn(
"TableId catalog is null/empty for table {}.{}, using database
name as fallback",
tableId.schema(),
tableId.table());
catalog = connectorConfig.databaseName();
if (catalog == null || catalog.isEmpty()) {
throw new IllegalStateException(
String.format(
"Cannot determine catalog for table %s.%s. " +
"Please check your PostgreSQL CDC configuration.",
tableId.schema(),
tableId.table()));
}
}
TableId newTableId = new TableId(catalog, tableId.schema(),
tableId.table());
```
Please add an E2E test for pg version 17 in
`seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-postgres-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/postgres/PostgresCDCIT.java`
, it will make sure PG cdc connector will work fine in these pg versions.
--
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]
