This is an automated email from the ASF dual-hosted git repository. leonardBang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit 4c266632e66c2fa4bd1804187a8bbee241d08d29 Author: Leonard Xu <[email protected]> AuthorDate: Wed Jul 1 14:57:44 2026 +0800 [test][connector/sqlserver] Stabilize full types and timezone waits --- .../connectors/sqlserver/SqlServerTestBase.java | 26 ++++++++++++----- .../sqlserver/table/SqlServerConnectorITCase.java | 33 ++++++++++++---------- .../sqlserver/table/SqlServerTimezoneITCase.java | 25 ++++++++++++++-- 3 files changed, 60 insertions(+), 24 deletions(-) diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java index 59ba390ce..0f52901c5 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/SqlServerTestBase.java @@ -59,6 +59,8 @@ public class SqlServerTestBase extends AbstractTestBase { private static final String STATEMENTS_PLACEHOLDER = "#"; + private static final long SINK_WAIT_TIMEOUT_SECONDS = 90; + private static final String DISABLE_DB_CDC = "IF EXISTS(select 1 from sys.databases where name='#' AND is_cdc_enabled=1)\n" + "EXEC sys.sp_cdc_disable_db"; @@ -175,6 +177,10 @@ public class SqlServerTestBase extends AbstractTestBase { * connection. */ protected void initializeSqlServerTable(String sqlFile) { + initializeSqlServerTable(sqlFile, sqlFile); + } + + protected void initializeSqlServerTable(String sqlFile, String databaseName) { final String ddlFile = String.format("ddl/%s.sql", sqlFile); final InputStream ddlTestFile = SqlServerTestBase.class.getClassLoader().getResourceAsStream(ddlFile); @@ -185,7 +191,7 @@ public class SqlServerTestBase extends AbstractTestBase { new InputStreamReader(inputStream, StandardCharsets.UTF_8)); Connection connection = getJdbcConnection(); Statement statement = connection.createStatement()) { - dropTestDatabase(connection, sqlFile); + dropTestDatabase(connection, databaseName); final List<String> statements = Arrays.stream( reader.lines() @@ -198,6 +204,7 @@ public class SqlServerTestBase extends AbstractTestBase { return m.matches() ? m.group(1) : x; }) .collect(Collectors.joining("\n")) + .replace(sqlFile, databaseName) .split(";")) .collect(Collectors.toList()); for (String stmt : statements) { @@ -209,16 +216,21 @@ public class SqlServerTestBase extends AbstractTestBase { } protected static void waitForSnapshotStarted(String sinkName) throws InterruptedException { - while (sinkSize(sinkName) == 0) { - Thread.sleep(100); - } + // Bound the wait so that a stalled source (e.g. a record that never converges because a + // streaming conversion failed) fails this single test fast instead of hanging the whole + // suite until the job-level timeout. + Awaitility.await(String.format("Snapshot of %s to start", sinkName)) + .atMost(SINK_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .until(() -> sinkSize(sinkName) > 0); } protected static void waitForSinkSize(String sinkName, int expectedSize) throws InterruptedException { - while (sinkSize(sinkName) < expectedSize) { - Thread.sleep(100); - } + Awaitility.await(String.format("Sink %s to reach size %d", sinkName, expectedSize)) + .atMost(SINK_WAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS) + .pollInterval(100, TimeUnit.MILLISECONDS) + .until(() -> sinkSize(sinkName) >= expectedSize); } protected static int sinkSize(String sinkName) { 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 ae1515f0c..48de0dc42 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 @@ -447,24 +447,27 @@ class SqlServerConnectorITCase extends SqlServerTestBase { // async submit job TableResult result = tEnv.executeSql("INSERT INTO sink SELECT * FROM full_types"); - waitForSnapshotStarted("sink"); - - try (Connection connection = getJdbcConnection(); - Statement statement = connection.createStatement()) { - statement.execute( - "UPDATE column_type_test.dbo.full_types SET val_int=8888 WHERE id=0;"); - } + try { + waitForSnapshotStarted("sink"); - waitForSinkSize("sink", 2); + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute( + "UPDATE column_type_test.dbo.full_types SET val_int=8888 WHERE id=0;"); + } - List<String> expected = - Arrays.asList( - "+I(0,cc ,vcc,tc,cč ,vcč,tč,1.123,2,3.323,4.323,5,6,true,22,333,4444,55555,2018-07-13,10:23:45.680,10:23:45.678,2018-07-13T11:23:45.340,2018-07-13T01:23:45.456Z,2018-07-13T13:23:45.780,2018-07-13T14:24,<a>b</a>)", - "+U(0,cc ,vcc,tc,cč ,vcč,tč,1.123,2,3.323,4.323,5,6,true,22,333,8888,55555,2018-07-13,10:23:45.680,10:23:45.679,2018-07-13T11:23:45.340,2018-07-13T01:23:45.456Z,2018-07-13T13:23:45.780,2018-07-13T14:24,<a>b</a>)"); - List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("sink"); - Assertions.assertThat(actual).containsExactlyInAnyOrderElementsOf(expected); + waitForSinkSize("sink", 2); - result.getJobClient().get().cancel().get(); + String expectedSnapshotRecord = + "+I(0,cc ,vcc,tc,cč ,vcč,tč,1.123,2,3.323,4.323,5,6,true,22,333,4444,55555,2018-07-13,10:23:45.680,10:23:45.678,2018-07-13T11:23:45.340,2018-07-13T01:23:45.456Z,2018-07-13T13:23:45.780,2018-07-13T14:24,<a>b</a>)"; + String expectedUpdatedRecord = + "+U(0,cc ,vcc,tc,cč ,vcč,tč,1.123,2,3.323,4.323,5,6,true,22,333,8888,55555,2018-07-13,10:23:45.680,10:23:45.679,2018-07-13T11:23:45.340,2018-07-13T01:23:45.456Z,2018-07-13T13:23:45.780,2018-07-13T14:24,<a>b</a>)"; + List<String> actual = TestValuesTableFactory.getRawResultsAsStrings("sink"); + Assertions.assertThat(actual) + .containsExactlyInAnyOrder(expectedSnapshotRecord, expectedUpdatedRecord); + } finally { + result.getJobClient().get().cancel().get(); + } } @ParameterizedTest diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTimezoneITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTimezoneITCase.java index efc092d35..f0c6ae839 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTimezoneITCase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/table/SqlServerTimezoneITCase.java @@ -124,7 +124,9 @@ class SqlServerTimezoneITCase extends SqlServerTestBase { TimeZone aDefault = TimeZone.getDefault(); try { TimeZone.setDefault(TimeZone.getTimeZone(jvmTimeZone)); - initializeSqlServerTable("column_type_test"); + String databaseName = + getDatabaseName(localTimeZone, serverTimeZone, jvmTimeZone, castTimeStampLtz); + initializeSqlServerTable("column_type_test", databaseName); String sourceDDL = String.format( @@ -151,7 +153,7 @@ class SqlServerTimezoneITCase extends SqlServerTestBase { MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT), MSSQL_SERVER_CONTAINER.getUsername(), MSSQL_SERVER_CONTAINER.getPassword(), - "column_type_test", + databaseName, "dbo.full_types", serverTimeZone); String sinkDDL = @@ -190,4 +192,23 @@ class SqlServerTimezoneITCase extends SqlServerTestBase { TimeZone.setDefault(aDefault); } } + + private String getDatabaseName( + String localTimeZone, + String serverTimeZone, + String jvmTimeZone, + boolean castTimeStampLtz) { + return "column_type_test_" + + normalizeTimeZoneName(localTimeZone) + + "_" + + normalizeTimeZoneName(serverTimeZone) + + "_" + + normalizeTimeZoneName(jvmTimeZone) + + "_" + + castTimeStampLtz; + } + + private String normalizeTimeZoneName(String timeZone) { + return timeZone.replaceAll("[^A-Za-z0-9]", "_"); + } }
