(flink-cdc) branch master updated: [FLINK-39231][test][sqlserver] Fix unstable sqlserver connector test (#4309)

[kunni]

This is an automated email from the ASF dual-hosted git repository.

kunni pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git


The following commit(s) were added to refs/heads/master by this push:
     new 368a8079b [FLINK-39231][test][sqlserver] Fix unstable sqlserver 
connector test (#4309)
368a8079b is described below

commit 368a8079bcb3541e74f4dd436447f94955857e3c
Author: Mingliang Zhu <zhuml1...@gmail.com>
AuthorDate: Thu Mar 12 10:07:10 2026 +0800

    [FLINK-39231][test][sqlserver] Fix unstable sqlserver connector test (#4309)
---
 .../sqlserver/table/SqlServerConnectorITCase.java  | 26 +++++++++++++---------
 1 file changed, 16 insertions(+), 10 deletions(-)

diff --git 
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerConnectorITCase.java
 
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerConnectorITCase.java
index e46defe44..25cb6ba59 100644
--- 
a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerConnectorITCase.java
+++ 
b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerConnectorITCase.java
@@ -72,6 +72,10 @@ class SqlServerConnectorITCase extends SqlServerTestBase {
         setup(parallelismSnapshot);
         initializeSqlServerTable("inventory");
         initializeSqlServerTable("product");
+        int expectedSize = 20;
+        if (parallelismSnapshot) {
+            expectedSize = 22;
+        }
         String sourceDDL =
                 String.format(
                         "CREATE TABLE debezium_source ("
@@ -98,15 +102,17 @@ class SqlServerConnectorITCase extends SqlServerTestBase {
                         "inventory",
                         "dbo.products");
         String sinkDDL =
-                "CREATE TABLE sink ("
-                        + " name STRING,"
-                        + " weightSum DECIMAL(10,3),"
-                        + " PRIMARY KEY (name) NOT ENFORCED"
-                        + ") WITH ("
-                        + " 'connector' = 'values',"
-                        + " 'sink-insert-only' = 'false',"
-                        + " 'sink-expected-messages-num' = '20'"
-                        + ")";
+                String.format(
+                        "CREATE TABLE sink ("
+                                + " name STRING,"
+                                + " weightSum DECIMAL(10,3),"
+                                + " PRIMARY KEY (name) NOT ENFORCED"
+                                + ") WITH ("
+                                + " 'connector' = 'values',"
+                                + " 'sink-insert-only' = 'false',"
+                                + " 'sink-expected-messages-num' = '%s'"
+                                + ")",
+                        expectedSize);
         tEnv.executeSql(sourceDDL);
         tEnv.executeSql(sinkDDL);
 
@@ -151,7 +157,7 @@ class SqlServerConnectorITCase extends SqlServerTestBase {
             statement.execute("UPDATE inventory.dbo.products SET volume='1.2' 
WHERE id=110;");
         }
 
-        waitForSinkSize("sink", 20);
+        waitForSinkSize("sink", expectedSize);
 
         /*
          * <pre>