uicosp commented on code in PR #3776:
URL: https://github.com/apache/flink-cdc/pull/3776#discussion_r4323953619
##########
flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/MysqlE2eITCase.java:
##########
@@ -535,4 +535,54 @@ void testDanglingDropTableEventInBinlog() throws Exception
{
"CreateTableEvent{tableId=%s.products, schema=columns={`id`
INT NOT NULL,`name` VARCHAR(255) NOT NULL 'flink',`description`
VARCHAR(512),`weight` FLOAT,`enum_c` STRING 'red',`json_c` STRING,`point_c`
STRING}, primaryKeys=id, options=()}",
"DataChangeEvent{tableId=%s.products, before=[106, hammer,
16oz carpenter's hammer, 1.0, null, null, null], after=[106, hammer, 18oz
carpenter hammer, 1.0, null, null, null], op=UPDATE, meta=()}");
}
+
+ @Test
+ void testSnapshotFilters() throws Exception {
+ String pipelineJob =
+ String.format(
+ "source:\n"
+ + " type: mysql\n"
+ + " hostname: %s\n"
+ + " port: 3306\n"
+ + " username: %s\n"
+ + " password: %s\n"
+ + " tables: %s.\\.*\n"
+ + " server-id: 5400-5404\n"
+ + " server-time-zone: UTC\n"
+ + " scan.snapshot.filters:\n"
+ + " - table: %s.customers\n"
+ + " filter: id > 102\n"
+ + " - table: %s.products\n"
+ + " filter: id < 105\n"
+ + "\n"
+ + "sink:\n"
+ + " type: values\n"
+ + "\n"
+ + "pipeline:\n"
+ + " parallelism: %d",
+ INTER_CONTAINER_MYSQL_ALIAS,
+ MYSQL_TEST_USER,
+ MYSQL_TEST_PASSWORD,
+ mysqlInventoryDatabase.getDatabaseName(),
+ mysqlInventoryDatabase.getDatabaseName(),
+ mysqlInventoryDatabase.getDatabaseName(),
+ parallelism);
+
+ submitPipelineJob(pipelineJob);
+ waitUntilJobRunning(Duration.ofSeconds(30));
+ LOG.info("Pipeline job is running");
+
+ // customers: only id > 102 (id=103, 104)
+ // products: only id < 105 (id=101, 102, 103, 104)
+ validateResult(
Review Comment:
Updated the E2E test to use bounded snapshot mode and explicitly verify that
the job reaches FINISHED. It then compares all emitted data events against the
complete expected result set, including the event count, using
containsExactlyInAnyOrderElementsOf().
This ensures that any additional rows cause the test to fail, including rows
that should be excluded by the filters. The test passes with parallelism 1 and
4.
--
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]
