phamvinh1712 commented on code in PR #4004:
URL: https://github.com/apache/flink-cdc/pull/4004#discussion_r2072688456
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/table/PostgreSQLConnectorITCase.java:
##########
@@ -232,6 +232,118 @@ void testConsumingAllEvents(boolean parallelismSnapshot)
result.getJobClient().get().cancel().get();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ void testConsumingAllEventsForPartitionedTable(boolean parallelismSnapshot)
+ throws SQLException, ExecutionException, InterruptedException {
+ setup(parallelismSnapshot);
+ initializePostgresTable(POSTGRES_CONTAINER, "inventory_partitioned");
+ String publicationName = "dbz_publication_" + new
Random().nextInt(1000);
+ String slotName = getSlotName();
+ try (Connection connection = getJdbcConnection(POSTGRES_CONTAINER);
+ Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "CREATE PUBLICATION %s FOR TABLE
inventory_partitioned.products "
+ + " WITH
(publish_via_partition_root=true)",
+ publicationName));
+ statement.execute(
+ String.format(
+ "select
pg_create_logical_replication_slot('%s','pgoutput');",
+ slotName));
+ }
+
+ String sourceDDL =
+ String.format(
+ "CREATE TABLE debezium_source ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " description STRING,"
+ + " weight DECIMAL(10,3),"
+ + " country STRING"
+ + ") WITH ("
+ + " 'connector' = 'postgres-cdc',"
+ + " 'hostname' = '%s',"
+ + " 'port' = '%s',"
+ + " 'username' = '%s',"
+ + " 'password' = '%s',"
+ + " 'database-name' = '%s',"
+ + " 'schema-name' = '%s',"
+ + " 'table-name' = '%s',"
+ + " 'scan.incremental.snapshot.enabled' =
'%s',"
+ + " 'decoding.plugin.name' = 'pgoutput', "
+ + " 'debezium.publication.name' = '%s',"
+ + " 'slot.name' = '%s'"
+ + ")",
+ POSTGRES_CONTAINER.getHost(),
+ POSTGRES_CONTAINER.getMappedPort(POSTGRESQL_PORT),
+ POSTGRES_CONTAINER.getUsername(),
+ POSTGRES_CONTAINER.getPassword(),
+ POSTGRES_CONTAINER.getDatabaseName(),
+ "inventory_partitioned",
+ "products",
+ parallelismSnapshot,
+ publicationName,
+ slotName);
+ String sinkDDL =
+ "CREATE TABLE sink ("
+ + " id INT NOT NULL,"
+ + " name STRING,"
+ + " description STRING,"
+ + " weight DECIMAL(10,3),"
+ + " country STRING,"
+ + " PRIMARY KEY (id, country) NOT ENFORCED"
+ + ") WITH ("
+ + " 'connector' = 'values',"
+ + " 'sink-insert-only' = 'false',"
+ + " 'sink-expected-messages-num' = '20'"
+ + ")";
+ tEnv.executeSql(sourceDDL);
+ tEnv.executeSql(sinkDDL);
+
+ // async submit job
+ TableResult result =
+ tEnv.executeSql(
+ "INSERT INTO sink SELECT id, name, description,
weight, country FROM debezium_source");
+
+ waitForSnapshotStarted("sink");
+
+ // wait a bit to make sure the replication slot is ready
+ Thread.sleep(5000);
+
+ // generate WAL
+ try (Connection connection = getJdbcConnection(POSTGRES_CONTAINER);
+ Statement statement = connection.createStatement()) {
Review Comment:
@loserwang1024 i've updated test for this.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
