Copilot commented on code in PR #4320: URL: https://github.com/apache/flink-cdc/pull/4320#discussion_r3042704327
########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerMetadataAccessor.java: ########## @@ -0,0 +1,84 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.source; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.source.MetadataAccessor; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig; +import org.apache.flink.cdc.connectors.sqlserver.utils.SqlServerSchemaUtils; + +import javax.annotation.Nullable; + +import java.util.List; + +/** {@link MetadataAccessor} for {@link SqlServerDataSource}. */ +@Internal +public class SqlServerMetadataAccessor implements MetadataAccessor { + + private final SqlServerSourceConfig sourceConfig; + + public SqlServerMetadataAccessor(SqlServerSourceConfig sourceConfig) { + this.sourceConfig = sourceConfig; + } + + /** + * List all databases from SQL Server. + * + * @return The list of database names + */ + @Override + public List<String> listNamespaces() { + return SqlServerSchemaUtils.listNamespaces(sourceConfig); + } + + /** + * List all schemas from SQL Server database. + * + * @param namespace The database name to list schemas from. + * @return The list of schema names + */ + @Override + public List<String> listSchemas(@Nullable String namespace) { + return SqlServerSchemaUtils.listSchemas(sourceConfig, namespace); + } + Review Comment: `listSchemas(@Nullable String namespace)` forwards a nullable value into `SqlServerSchemaUtils.listSchemas(...)`, which currently formats SQL using `quote(namespace)` and will NPE (or query `[null]`). Handle `namespace == null` explicitly (e.g., default to the configured database from `sourceConfig.getDatabaseList()` or throw a `ValidationException` with a clear message). ```suggestion return SqlServerSchemaUtils.listSchemas(sourceConfig, resolveNamespace(namespace)); } private String resolveNamespace(@Nullable String namespace) { if (namespace != null) { return namespace; } List<String> configuredDatabases = sourceConfig.getDatabaseList(); if (configuredDatabases != null && !configuredDatabases.isEmpty()) { return configuredDatabases.get(0); } throw new IllegalArgumentException( "Namespace must not be null when listing SQL Server schemas and no " + "database is configured in the source configuration."); } ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/resources/ddl/inventory.sql: ########## @@ -0,0 +1,106 @@ +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +-- ---------------------------------------------------------------------------------------------------------------- +-- DATABASE: inventory +-- ---------------------------------------------------------------------------------------------------------------- +-- Create the inventory database +CREATE DATABASE inventory; + +USE inventory; + +-- Avoid SqlServer error com.microsoft.sqlserver.jdbc.SQLServerException: Could not update the metadata that indicates, +-- the root cause is 14258: 'Cannot perform this operation while SQLServerAgent is starting. Try again later.'. We simply +-- wait for 3 seconds to improve the test stabilises. +WAITFOR DELAY '00:00:03'; +EXEC sys.sp_cdc_enable_db; + +-- Create and populate our products using a single insert with many rows +CREATE TABLE products ( + id INTEGER IDENTITY(101,1) NOT NULL PRIMARY KEY, + name VARCHAR(255) NOT NULL, + description VARCHAR(512), + weight FLOAT +); +INSERT INTO products(name,description,weight) +VALUES ('scooter','Small 2-wheel scooter',3.14); +INSERT INTO products(name,description,weight) +VALUES ('car battery','12V car battery',8.1); +INSERT INTO products(name,description,weight) +VALUES ('12-pack drill bits','12-pack of drill bits with sizes ranging from #40 to #3',0.8); +INSERT INTO products(name,description,weight) +VALUES ('hammer','12oz carpenter''s hammer',0.75); +INSERT INTO products(name,description,weight) +VALUES ('hammer','14oz carpenter''s hammer',0.875); +INSERT INTO products(name,description,weight) +VALUES ('hammer','16oz carpenter''s hammer',1.0); +INSERT INTO products(name,description,weight) +VALUES ('rocks','box of assorted rocks',5.3); +INSERT INTO products(name,description,weight) +VALUES ('jacket','water resistent black wind breaker',0.1); Review Comment: Typo in sample data: 'resistent' should be 'resistant'. If any assertions depend on the exact string, update those expected values accordingly. ```suggestion VALUES ('jacket','water resistant black wind breaker',0.1); ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerPipelineNewlyAddedTableITCase.java: ########## @@ -0,0 +1,666 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.source; + +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.restartstrategy.RestartStrategies; +import org.apache.flink.api.common.typeutils.TypeSerializer; +import org.apache.flink.cdc.common.data.binary.BinaryStringData; +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.factories.Factory; +import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.source.FlinkSourceProvider; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.common.types.DataTypes; +import org.apache.flink.cdc.common.types.RowType; +import org.apache.flink.cdc.connectors.sqlserver.SqlServerTestBase; +import org.apache.flink.cdc.connectors.sqlserver.factory.SqlServerDataSourceFactory; +import org.apache.flink.cdc.connectors.sqlserver.testutils.SqlServerSourceTestUtils; +import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator; +import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.core.execution.JobClient; +import org.apache.flink.runtime.checkpoint.CheckpointException; +import org.apache.flink.runtime.jobgraph.SavepointConfigOptions; +import org.apache.flink.streaming.api.datastream.DataStreamSource; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.streaming.api.operators.collect.AbstractCollectResultBuffer; +import org.apache.flink.streaming.api.operators.collect.CheckpointedCollectResultBuffer; +import org.apache.flink.streaming.api.operators.collect.CollectResultIterator; +import org.apache.flink.streaming.api.operators.collect.CollectSinkOperator; +import org.apache.flink.streaming.api.operators.collect.CollectSinkOperatorFactory; +import org.apache.flink.streaming.api.operators.collect.CollectStreamSink; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.util.ExceptionUtils; + +import org.apache.commons.collections.CollectionUtils; +import org.apache.commons.lang3.StringUtils; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.nio.file.Files; +import java.nio.file.Path; +import java.sql.Connection; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.HOSTNAME; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PASSWORD; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PORT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SERVER_TIME_ZONE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES_EXCLUDE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.USERNAME; +import static org.assertj.core.api.Assertions.assertThat; +import static org.testcontainers.containers.MSSQLServerContainer.MS_SQL_SERVER_PORT; + +/** IT tests to cover various newly added tables during capture process in pipeline mode. */ +class SqlServerPipelineNewlyAddedTableITCase extends SqlServerTestBase { + + private static final String DATABASE_NAME = "newly_added_table_test"; + private static final String SCHEMA_NAME = "dbo"; + private static final int DEFAULT_PARALLELISM = 4; + + private final ScheduledExecutorService mockCdcExecutor = Executors.newScheduledThreadPool(1); + + @BeforeEach + void before() throws SQLException { + TestValuesTableFactory.clearAllData(); + initializeDatabase(); + } + + @AfterEach + void after() { + mockCdcExecutor.shutdown(); + } + + private void initializeDatabase() throws SQLException { + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + // Drop database if exists + statement.execute( + String.format( + "IF EXISTS(select 1 from sys.databases where name = '%s') " + + "BEGIN ALTER DATABASE [%s] SET SINGLE_USER WITH ROLLBACK IMMEDIATE; " + + "DROP DATABASE [%s]; END", + DATABASE_NAME, DATABASE_NAME, DATABASE_NAME)); + + // Create database + statement.execute(String.format("CREATE DATABASE %s;", DATABASE_NAME)); + statement.execute(String.format("USE %s;", DATABASE_NAME)); + + // Wait for SQL Server Agent + statement.execute("WAITFOR DELAY '00:00:03';"); + statement.execute("EXEC sys.sp_cdc_enable_db;"); + + // Create produce_cdc_table for background CDC activity + statement.execute( + "CREATE TABLE dbo.produce_cdc_table (id BIGINT NOT NULL PRIMARY KEY, cnt BIGINT);"); + statement.execute( + "INSERT INTO dbo.produce_cdc_table VALUES (0, 100), (1, 101), (2, 102);"); + statement.execute( + "EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', " + + "@source_name = 'produce_cdc_table', @role_name = NULL, @supports_net_changes = 0;"); + + // Mock continuous CDC during the newly added table capturing process + mockCdcExecutor.schedule( + () -> { + try (Connection conn = getJdbcConnection(); + Statement stmt = conn.createStatement()) { + stmt.execute(String.format("USE %s;", DATABASE_NAME)); + stmt.execute( + "UPDATE dbo.produce_cdc_table SET cnt = cnt + 1 WHERE id < 2;"); + } catch (SQLException e) { + e.printStackTrace(); + } + }, + 500, + TimeUnit.MILLISECONDS); + } + } + + @Test + void testAddNewTableOneByOneSingleParallelism() throws Exception { + TestParam testParam = + TestParam.newBuilder( + Collections.singletonList("address_hangzhou"), + 4, + Arrays.asList("address_hangzhou", "address_beijing"), + 4) + .setFirstRoundInitTables( + Arrays.asList("address_hangzhou", "address_beijing")) + .build(); + + testAddNewTable(testParam, 1); + } + + @Test + void testAddNewTableOneByOne() throws Exception { + TestParam testParam = + TestParam.newBuilder( + Collections.singletonList("address_hangzhou"), + 4, + Arrays.asList("address_hangzhou", "address_beijing"), + 4) + .setFirstRoundInitTables( + Arrays.asList("address_hangzhou", "address_beijing")) + .build(); + + testAddNewTable(testParam, DEFAULT_PARALLELISM); + } + + @Test + void testAddNewTableByPatternSingleParallelism() throws Exception { + TestParam testParam = + TestParam.newBuilder( + Collections.singletonList("address_\\.*"), + 8, + Collections.singletonList("address_\\.*"), + 8) + .setFirstRoundInitTables( + Arrays.asList("address_hangzhou", "address_beijing")) + .setSecondRoundInitTables( + Arrays.asList("address_shanghai", "address_suzhou")) + .build(); + + testAddNewTable(testParam, 1); + } + + @Test + void testAddNewTableByPattern() throws Exception { + TestParam testParam = + TestParam.newBuilder( + Collections.singletonList("address_\\.*"), + 8, + Collections.singletonList("address_\\.*"), + 12) + .setFirstRoundInitTables( + Arrays.asList("address_hangzhou", "address_beijing")) + .setSecondRoundInitTables( + Arrays.asList( + "address_shanghai", "address_suzhou", "address_shenzhen")) + .build(); + + testAddNewTable(testParam, DEFAULT_PARALLELISM); + } + + @Test + void testAddNewTableWithExclude() throws Exception { + // Initialize all tables first + initialAddressTables( + Arrays.asList("address_hangzhou", "address_beijing", "address_excluded")); + + Map<String, String> options = new HashMap<>(); + options.put(TABLES_EXCLUDE.key(), DATABASE_NAME + "." + SCHEMA_NAME + ".address_excluded"); + + FlinkSourceProvider sourceProvider = + getFlinkSourceProvider( + Collections.singletonList("address_\\.*"), + DEFAULT_PARALLELISM, + options, + true); + StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(new Configuration()); + env.enableCheckpointing(200); + DataStreamSource<Event> source = + env.fromSource( + sourceProvider.getSource(), + WatermarkStrategy.noWatermarks(), + SqlServerDataSourceFactory.IDENTIFIER, + new EventTypeInfo()); + + TypeSerializer<Event> serializer = + source.getTransformation().getOutputType().createSerializer(env.getConfig()); + CheckpointedCollectResultBuffer<Event> resultBuffer = + new CheckpointedCollectResultBuffer<>(serializer); + String accumulatorName = "dataStreamCollect_" + UUID.randomUUID(); + CollectResultIterator<Event> iterator = + addCollector(env, source, resultBuffer, serializer, accumulatorName); + JobClient jobClient = env.executeAsync("TestExcludeTables"); + iterator.setJobClient(jobClient); + + // Expect 2 tables (hangzhou, beijing), each with CreateTableEvent + 3 DataChangeEvents = 8 + // total + List<Event> actual = SqlServerSourceTestUtils.fetchResults(iterator, 8); + + List<String> tableNames = + actual.stream() + .filter(event -> event instanceof CreateTableEvent) + .map(event -> ((SchemaChangeEvent) event).tableId().getTableName()) + .collect(Collectors.toList()); + + assertThat(tableNames).hasSize(2); + assertThat(tableNames).containsExactlyInAnyOrder("address_hangzhou", "address_beijing"); + assertThat(tableNames).doesNotContain("address_excluded"); + + jobClient.cancel().get(); + iterator.close(); + } + + private void testAddNewTable(TestParam testParam, int parallelism) throws Exception { + // step 1: create sqlserver tables + if (CollectionUtils.isNotEmpty(testParam.getFirstRoundInitTables())) { + initialAddressTables(testParam.getFirstRoundInitTables()); + } + Path savepointDir = Files.createTempDirectory("add-new-table-test"); + final String savepointDirectory = savepointDir.toAbsolutePath().toString(); + String finishedSavePointPath = null; + StreamExecutionEnvironment env = + getStreamExecutionEnvironment(finishedSavePointPath, parallelism); + + // step 2: listen tables first time + List<String> listenTablesFirstRound = testParam.getFirstRoundListenTables(); + + FlinkSourceProvider sourceProvider = + getFlinkSourceProvider(listenTablesFirstRound, parallelism, new HashMap<>(), true); + DataStreamSource<Event> source = + env.fromSource( + sourceProvider.getSource(), + WatermarkStrategy.noWatermarks(), + SqlServerDataSourceFactory.IDENTIFIER, + new EventTypeInfo()); + + TypeSerializer<Event> serializer = + source.getTransformation().getOutputType().createSerializer(env.getConfig()); + CheckpointedCollectResultBuffer<Event> resultBuffer = + new CheckpointedCollectResultBuffer<>(serializer); + String accumulatorName = "dataStreamCollect_" + UUID.randomUUID(); + CollectResultIterator<Event> iterator = + addCollector(env, source, resultBuffer, serializer, accumulatorName); + JobClient jobClient = env.executeAsync("beforeAddNewTable"); + iterator.setJobClient(jobClient); + + List<Event> actual = + SqlServerSourceTestUtils.fetchResults(iterator, testParam.getFirstRoundFetchSize()); + Optional<String> listenByPattern = + listenTablesFirstRound.stream() + .filter(table -> StringUtils.contains(table, "\\.*")) + .findAny(); + multiAssert( + actual, + listenByPattern.isPresent() + ? testParam.getFirstRoundInitTables() + : listenTablesFirstRound); + + // step 3: create new tables if needed + if (CollectionUtils.isNotEmpty(testParam.getSecondRoundInitTables())) { + initialAddressTables(testParam.getSecondRoundInitTables()); + } + + // sleep 1s to wait for the assign status to INITIAL_ASSIGNING_FINISHED. + // Otherwise, the restart job won't read newly added tables, and this test will be stuck. + Thread.sleep(1000L); + // step 4: trigger a savepoint and cancel the job + finishedSavePointPath = triggerSavepointWithRetry(jobClient, savepointDirectory); + jobClient.cancel().get(); + iterator.close(); + + // step 5: restore from savepoint + StreamExecutionEnvironment restoredEnv = + getStreamExecutionEnvironment(finishedSavePointPath, parallelism); + List<String> listenTablesSecondRound = testParam.getSecondRoundListenTables(); + FlinkSourceProvider restoredSourceProvider = + getFlinkSourceProvider(listenTablesSecondRound, parallelism, new HashMap<>(), true); + DataStreamSource<Event> restoreSource = + restoredEnv.fromSource( + restoredSourceProvider.getSource(), + WatermarkStrategy.noWatermarks(), + SqlServerDataSourceFactory.IDENTIFIER, + new EventTypeInfo()); + CollectResultIterator<Event> restoredIterator = + addCollector(restoredEnv, restoreSource, resultBuffer, serializer, accumulatorName); + JobClient restoreClient = restoredEnv.executeAsync("AfterAddNewTable"); + + List<String> newlyAddTables = + listenTablesSecondRound.stream() + .filter(table -> !listenTablesFirstRound.contains(table)) + .collect(Collectors.toList()); + // it means listen by pattern when newlyAddTables is empty + if (CollectionUtils.isEmpty(newlyAddTables)) { + newlyAddTables = testParam.getSecondRoundInitTables(); + } + List<Event> newlyTableEvent = + SqlServerSourceTestUtils.fetchResults( + restoredIterator, testParam.getSecondRoundFetchSize()); + multiAssert(newlyTableEvent, newlyAddTables); + restoreClient.cancel().get(); + restoredIterator.close(); + } + + private void multiAssert(List<Event> actualEvents, List<String> listenTables) { + List<Event> expectedCreateTableEvents = new ArrayList<>(); + List<Event> expectedDataChangeEvents = new ArrayList<>(); + for (String table : listenTables) { + expectedCreateTableEvents.add( + getCreateTableEvent(TableId.tableId(DATABASE_NAME, SCHEMA_NAME, table))); + expectedDataChangeEvents.addAll( + getSnapshotExpected(TableId.tableId(DATABASE_NAME, SCHEMA_NAME, table))); + } + // compare create table events + List<Event> actualCreateTableEvents = + actualEvents.stream() + .filter(event -> event instanceof CreateTableEvent) + .collect(Collectors.toList()); + assertThat(actualCreateTableEvents) + .containsExactlyInAnyOrder(expectedCreateTableEvents.toArray(new Event[0])); + + // compare data change events + List<Event> actualDataChangeEvents = + actualEvents.stream() + .filter(event -> event instanceof DataChangeEvent) + .collect(Collectors.toList()); + assertThat(actualDataChangeEvents) + .containsExactlyInAnyOrder(expectedDataChangeEvents.toArray(new Event[0])); + } + + private CreateTableEvent getCreateTableEvent(TableId tableId) { + Schema schema = + Schema.newBuilder() + .physicalColumn("id", DataTypes.BIGINT().notNull()) + .physicalColumn("country", DataTypes.VARCHAR(255).notNull()) + .physicalColumn("city", DataTypes.VARCHAR(255).notNull()) + .physicalColumn("detail_address", DataTypes.VARCHAR(1024)) + .primaryKey(Collections.singletonList("id")) + .build(); + return new CreateTableEvent(tableId, schema); + } + + private List<Event> getSnapshotExpected(TableId tableId) { + RowType rowType = + RowType.of( + new DataType[] { + DataTypes.BIGINT().notNull(), + DataTypes.VARCHAR(255).notNull(), + DataTypes.VARCHAR(255).notNull(), + DataTypes.VARCHAR(1024) + }, + new String[] {"id", "country", "city", "detail_address"}); + BinaryRecordDataGenerator generator = new BinaryRecordDataGenerator(rowType); + String cityName = tableId.getTableName().split("_")[1]; + return Arrays.asList( + DataChangeEvent.insertEvent( + tableId, + generator.generate( + new Object[] { + 416874195632735147L, + BinaryStringData.fromString("China"), + BinaryStringData.fromString(cityName), + BinaryStringData.fromString(cityName + " West Town address 1") + })), + DataChangeEvent.insertEvent( + tableId, + generator.generate( + new Object[] { + 416927583791428523L, + BinaryStringData.fromString("China"), + BinaryStringData.fromString(cityName), + BinaryStringData.fromString(cityName + " West Town address 2") + })), + DataChangeEvent.insertEvent( + tableId, + generator.generate( + new Object[] { + 417022095255614379L, + BinaryStringData.fromString("China"), + BinaryStringData.fromString(cityName), + BinaryStringData.fromString(cityName + " West Town address 3") + }))); + } + + private String triggerSavepointWithRetry(JobClient jobClient, String savepointDirectory) + throws ExecutionException, InterruptedException { + int retryTimes = 0; + // retry 600 times, it takes 100 milliseconds per time, at most retry 1 minute + while (retryTimes < 600) { + try { + return jobClient.triggerSavepoint(savepointDirectory).get(); + } catch (Exception e) { + Optional<CheckpointException> exception = + ExceptionUtils.findThrowable(e, CheckpointException.class); + if (exception.isPresent() + && exception.get().getMessage().contains("Checkpoint triggering task")) { + Thread.sleep(100); + retryTimes++; + } else { + throw e; + } + } + } + return null; Review Comment: If the savepoint never succeeds after retries, the method returns `null`, and the test proceeds as if it had a savepoint path. This can lead to misleading failures (or restoring without a savepoint). Prefer throwing an exception (e.g., `TimeoutException`/`AssertionError`) once retries are exhausted so the test fails fast with a clear reason. ```suggestion throw new AssertionError( String.format( "Failed to trigger savepoint in directory '%s' after %d retries.", savepointDirectory, retryTimes)); ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/utils/SqlServerTypeUtils.java: ########## @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.utils; + +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.common.types.DataTypes; +import org.apache.flink.table.types.logical.DecimalType; + +import io.debezium.relational.Column; + +import java.sql.Types; + +/** A utility class for converting SQL Server types to Flink CDC types. */ +public class SqlServerTypeUtils { + + // SQL Server specific type names + static final String UNIQUE_IDENTIFIER = "uniqueidentifier"; + static final String XML = "xml"; + static final String SQL_VARIANT = "sql_variant"; + static final String HIERARCHY_ID = "hierarchyid"; + static final String GEOMETRY = "geometry"; + static final String GEOGRAPHY = "geography"; + static final String MONEY = "money"; + static final String SMALL_MONEY = "smallmoney"; + static final String DATETIME_OFFSET = "datetimeoffset"; + static final String DATETIME2 = "datetime2"; + static final String DATETIME = "datetime"; + static final String SMALL_DATETIME = "smalldatetime"; + static final String IMAGE = "image"; + static final String TIMESTAMP = "timestamp"; + static final String ROW_VERSION = "rowversion"; + static final String TEXT = "text"; + static final String N_TEXT = "ntext"; + + /** Returns a corresponding Flink CDC data type from a debezium {@link Column}. */ + public static DataType fromDbzColumn(Column column) { + DataType dataType = convertFromColumn(column); + if (column.isOptional()) { + return dataType; + } else { + return dataType.notNull(); + } + } + + /** + * Returns a corresponding Flink CDC data type from a debezium {@link Column} with nullable + * always be true. + */ + private static DataType convertFromColumn(Column column) { + int precision = column.length(); + int scale = column.scale().orElse(0); + + switch (column.jdbcType()) { + case Types.BIT: + case Types.BOOLEAN: + return DataTypes.BOOLEAN(); + case Types.TINYINT: + // SQL Server TINYINT is unsigned 0-255, maps to SMALLINT + return DataTypes.SMALLINT(); + case Types.SMALLINT: + return DataTypes.SMALLINT(); + case Types.INTEGER: + return DataTypes.INT(); + case Types.BIGINT: + return DataTypes.BIGINT(); + case Types.REAL: + case Types.FLOAT: + return DataTypes.FLOAT(); Review Comment: JDBC `Types.FLOAT` typically represents a double-precision floating type (while `Types.REAL` is single-precision). Mapping both `REAL` and `FLOAT` to `DataTypes.FLOAT()` can lose precision for SQL Server `FLOAT` columns and contradicts the expected behavior in the added type IT coverage. Consider mapping `Types.REAL -> DataTypes.FLOAT()` and `Types.FLOAT -> DataTypes.DOUBLE()` (or use Debezium column metadata to decide). ```suggestion return DataTypes.FLOAT(); case Types.FLOAT: ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactory.java: ########## @@ -0,0 +1,442 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.factory; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.factories.DataSourceFactory; +import org.apache.flink.cdc.common.factories.Factory; +import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Selectors; +import org.apache.flink.cdc.common.source.DataSource; +import org.apache.flink.cdc.common.utils.StringUtils; +import org.apache.flink.cdc.connectors.base.options.SourceOptions; +import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSource; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfigFactory; +import org.apache.flink.cdc.connectors.sqlserver.table.SqlServerReadableMetadata; +import org.apache.flink.cdc.connectors.sqlserver.utils.SqlServerSchemaUtils; +import org.apache.flink.table.api.ValidationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Properties; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_META_GROUP_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECTION_POOL_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_MAX_RETRIES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_TIMEOUT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.HOSTNAME; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.METADATA_LIST; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PASSWORD; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PORT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_MODE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCHEMA_CHANGE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SERVER_TIME_ZONE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES_EXCLUDE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.USERNAME; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.getDebeziumProperties; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.PROPERTIES_PREFIX; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.getJdbcProperties; +import static org.apache.flink.util.Preconditions.checkState; + +/** A {@link Factory} to create {@link SqlServerDataSource}. */ +@Internal +public class SqlServerDataSourceFactory implements DataSourceFactory { + + private static final Logger LOG = LoggerFactory.getLogger(SqlServerDataSourceFactory.class); + + public static final String IDENTIFIER = "sqlserver"; + + @Override + public DataSource createDataSource(Context context) { + FactoryHelper.createFactoryHelper(this, context) + .validateExcept(PROPERTIES_PREFIX, DEBEZIUM_OPTIONS_PREFIX); + + final Configuration config = context.getFactoryConfiguration(); + String hostname = config.get(HOSTNAME); + int port = config.get(PORT); + String username = config.get(USERNAME); + String password = config.get(PASSWORD); + String chunkKeyColumn = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN); + String tables = config.get(TABLES); + ZoneId serverTimeZone = getServerTimeZone(config); + String tablesExclude = config.get(TABLES_EXCLUDE); + StartupOptions startupOptions = getStartupOptions(config); + + boolean includeSchemaChanges = config.get(SCHEMA_CHANGE_ENABLED); + + int fetchSize = config.get(SCAN_SNAPSHOT_FETCH_SIZE); + int splitSize = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE); + int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE); + + double distributionFactorUpper = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND); + double distributionFactorLower = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); + + boolean closeIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); + boolean isAssignUnboundedChunkFirst = + config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED); + + Duration connectTimeout = config.get(CONNECT_TIMEOUT); + int connectMaxRetries = config.get(CONNECT_MAX_RETRIES); + int connectionPoolSize = config.get(CONNECTION_POOL_SIZE); + boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); + + validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1); + validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1); + validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1); + validateIntegerOption(CONNECTION_POOL_SIZE, connectionPoolSize, 1); + validateIntegerOption(CONNECT_MAX_RETRIES, connectMaxRetries, 0); + validateDistributionFactorUpper(distributionFactorUpper); + validateDistributionFactorLower(distributionFactorLower); + + Map<String, String> configMap = config.toMap(); + mergeJdbcPropertiesIntoDebeziumProperties(configMap); + String databaseName = getValidateDatabaseName(tables).orElseThrow(); + + SqlServerSourceConfigFactory configFactory = new SqlServerSourceConfigFactory(); + configFactory + .hostname(hostname) + .port(port) + .databaseList(databaseName) + .tableList(".*\\..*") + .username(username) + .password(password) + .serverTimeZone(serverTimeZone.getId()) + .debeziumProperties(getDebeziumProperties(configMap)) + .splitSize(splitSize) + .splitMetaGroupSize(splitMetaGroupSize) + .distributionFactorUpper(distributionFactorUpper) + .distributionFactorLower(distributionFactorLower) + .fetchSize(fetchSize) + .connectTimeout(connectTimeout) + .connectMaxRetries(connectMaxRetries) + .connectionPoolSize(connectionPoolSize) + .includeSchemaChanges(includeSchemaChanges) + .startupOptions(startupOptions) + .chunkKeyColumn(chunkKeyColumn) + .closeIdleReaders(closeIdleReaders) + .skipSnapshotBackfill(skipSnapshotBackfill) + .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled) + .assignUnboundedChunkFirst(isAssignUnboundedChunkFirst); + + SqlServerSourceConfig sourceConfig = configFactory.create(0); + SqlServerSchemaUtils.validateSqlServerAgentRunning(sourceConfig); + + List<TableId> tableIds = SqlServerSchemaUtils.listTables(sourceConfig, null, null); + + Selectors selectors = new Selectors.SelectorsBuilder().includeTables(tables).build(); + List<String> capturedTables = getTableList(tableIds, selectors); + if (capturedTables.isEmpty()) { + throw new IllegalArgumentException( + "Cannot find any table by the option 'tables' = " + tables); + } + if (tablesExclude != null) { + Selectors selectExclude = + new Selectors.SelectorsBuilder().includeTables(tablesExclude).build(); + List<String> excludeTables = getTableList(tableIds, selectExclude); + if (!excludeTables.isEmpty()) { + capturedTables.removeAll(excludeTables); + } + if (capturedTables.isEmpty()) { + throw new IllegalArgumentException( + "Cannot find any table with by the option 'tables.exclude' = " + + tablesExclude); + } + } + configFactory.tableList(capturedTables.toArray(new String[0])); + + String metadataList = config.get(METADATA_LIST); + List<SqlServerReadableMetadata> readableMetadataList = listReadableMetadata(metadataList); + + return new SqlServerDataSource(configFactory, readableMetadataList); + } + + private void mergeJdbcPropertiesIntoDebeziumProperties(Map<String, String> configMap) { + Properties jdbcProperties = getJdbcProperties(configMap); + for (String propertyName : jdbcProperties.stringPropertyNames()) { + configMap.putIfAbsent( + DEBEZIUM_OPTIONS_PREFIX + "database." + propertyName, + jdbcProperties.getProperty(propertyName)); + } + } + + private List<SqlServerReadableMetadata> listReadableMetadata(String metadataList) { + if (StringUtils.isNullOrWhitespaceOnly(metadataList)) { + return new ArrayList<>(); + } + Set<String> readableMetadataSet = + Arrays.stream(metadataList.split(",")) + .map(String::trim) + .collect(Collectors.toSet()); + List<SqlServerReadableMetadata> foundMetadata = new ArrayList<>(); + for (SqlServerReadableMetadata metadata : SqlServerReadableMetadata.values()) { + if (readableMetadataSet.contains(metadata.getKey())) { + foundMetadata.add(metadata); + readableMetadataSet.remove(metadata.getKey()); + } + } + if (readableMetadataSet.isEmpty()) { + return foundMetadata; + } + throw new IllegalArgumentException( + String.format( + "[%s] cannot be found in SQL Server metadata.", + String.join(", ", readableMetadataSet))); + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(HOSTNAME); + options.add(USERNAME); + options.add(PASSWORD); + options.add(TABLES); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PORT); + options.add(TABLES_EXCLUDE); + options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE); + options.add(SCAN_SNAPSHOT_FETCH_SIZE); + options.add(SCHEMA_CHANGE_ENABLED); + options.add(SCAN_STARTUP_MODE); + options.add(SCAN_STARTUP_TIMESTAMP_MILLIS); + options.add(SERVER_TIME_ZONE); + options.add(CONNECT_TIMEOUT); + options.add(CONNECT_MAX_RETRIES); + options.add(CONNECTION_POOL_SIZE); + options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); + options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); + options.add(CHUNK_META_GROUP_SIZE); + options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND); + options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); + options.add(METADATA_LIST); + options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED); + return options; + } + + @Override + public String identifier() { + return IDENTIFIER; + } + + private static List<String> getTableList( + @Nullable List<TableId> tableIdList, Selectors selectors) { + return Objects.requireNonNullElse(tableIdList, Collections.<TableId>emptyList()).stream() + .filter(selectors::isMatch) + // SQL Server tableList format: schemaName.tableName (without database prefix) + // See SqlServerSourceBuilder: "Each identifier is of the form + // <schemaName>.<tableName>" + .map(tableId -> tableId.getSchemaName() + "." + tableId.getTableName()) + .collect(Collectors.toList()); + } + + /** Checks the value of given integer option is valid. */ + private void validateIntegerOption( + ConfigOption<Integer> option, int optionValue, int exclusiveMin) { + checkState( + optionValue > exclusiveMin, + String.format( + "The value of option '%s' must larger than %d, but is %d", + option.key(), exclusiveMin, optionValue)); + } + + private static final String SCAN_STARTUP_MODE_VALUE_INITIAL = "initial"; + + private static final String SCAN_STARTUP_MODE_VALUE_SNAPSHOT = "snapshot"; + private static final String SCAN_STARTUP_MODE_VALUE_LATEST = "latest-offset"; + private static final String SCAN_STARTUP_MODE_VALUE_TIMESTAMP = "timestamp"; + + private static StartupOptions getStartupOptions(Configuration config) { + String modeString = config.get(SCAN_STARTUP_MODE); + + switch (modeString.toLowerCase()) { + case SCAN_STARTUP_MODE_VALUE_INITIAL: + return StartupOptions.initial(); + case SCAN_STARTUP_MODE_VALUE_SNAPSHOT: + return StartupOptions.snapshot(); + case SCAN_STARTUP_MODE_VALUE_LATEST: + return StartupOptions.latest(); + + case SCAN_STARTUP_MODE_VALUE_TIMESTAMP: + Long startupTimestampMillis = config.get(SCAN_STARTUP_TIMESTAMP_MILLIS); + if (startupTimestampMillis == null) { + throw new ValidationException( + String.format( + "To use timestamp startup mode, the startup timestamp millis '%s' must be set.", + SCAN_STARTUP_TIMESTAMP_MILLIS.key())); + } + return StartupOptions.timestamp(startupTimestampMillis); + + default: + throw new ValidationException( + String.format( + "Invalid value for option '%s'. Supported values are [%s, %s, %s, %s], but was: %s", + SourceOptions.SCAN_STARTUP_MODE.key(), + SCAN_STARTUP_MODE_VALUE_INITIAL, + SCAN_STARTUP_MODE_VALUE_SNAPSHOT, + SCAN_STARTUP_MODE_VALUE_LATEST, + SCAN_STARTUP_MODE_VALUE_TIMESTAMP, + modeString)); + } + } + + /** Checks the value of given evenly distribution factor upper bound is valid. */ + private void validateDistributionFactorUpper(double distributionFactorUpper) { + checkState( + doubleCompare(distributionFactorUpper, 1.0d) >= 0, + String.format( + "The value of option '%s' must larger than or equals %s, but is %s", + SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.key(), + 1.0d, + distributionFactorUpper)); + } + + /** Checks the value of given evenly distribution factor lower bound is valid. */ + private void validateDistributionFactorLower(double distributionFactorLower) { + checkState( + doubleCompare(distributionFactorLower, 0.0d) >= 0 + && doubleCompare(distributionFactorLower, 1.0d) <= 0, + String.format( + "The value of option '%s' must between %s and %s inclusively, but is %s", + SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.key(), + 0.0d, + 1.0d, + distributionFactorLower)); + } + + /** + * Get the database name from tables configuration. + * + * @param tables Table name list, format is "db.schema.table,db.schema.table,..." Each table + * name consists of three parts separated by ".", which are database name, schema name, and + * table name. + * @return Database name if found, otherwise returns Optional.empty() + * @throws IllegalArgumentException If the input parameter is null or does not match the + * expected format, or if database names are inconsistent. + */ + private Optional<String> getValidateDatabaseName(String tables) { + // Input validation + if (tables == null || tables.trim().isEmpty()) { + throw new IllegalArgumentException("Parameter tables cannot be null or empty"); + } + + // Split table name list + String[] tableNames = tables.split(","); + String dbName = null; + + for (String tableName : tableNames) { + // Trim whitespace and split table name + String trimmedTableName = tableName.trim(); + if (!trimmedTableName.contains(".")) { + continue; // Skip table names that do not match the expected format + } Review Comment: Two issues in database-name extraction/validation:\n1) If all entries are skipped (e.g., due to format issues), `getValidateDatabaseName(...)` can return `Optional.empty()`, and the caller uses `.orElseThrow()` without a message, resulting in a `NoSuchElementException` that doesn't explain the config problem.\n2) The length validation error message uses `TABLES.key()` (the literal key name) instead of the invalid value (e.g., `trimmedTableName` or the full `tables` string), making troubleshooting harder.\nRecommendation: treat any non-`db.schema.table` entry as a configuration error (fail fast with `ValidationException`/`IllegalArgumentException` containing the actual option value) and avoid returning empty when the option is required. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactory.java: ########## @@ -0,0 +1,442 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.factory; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.factories.DataSourceFactory; +import org.apache.flink.cdc.common.factories.Factory; +import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Selectors; +import org.apache.flink.cdc.common.source.DataSource; +import org.apache.flink.cdc.common.utils.StringUtils; +import org.apache.flink.cdc.connectors.base.options.SourceOptions; +import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSource; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfigFactory; +import org.apache.flink.cdc.connectors.sqlserver.table.SqlServerReadableMetadata; +import org.apache.flink.cdc.connectors.sqlserver.utils.SqlServerSchemaUtils; +import org.apache.flink.table.api.ValidationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Properties; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_META_GROUP_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECTION_POOL_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_MAX_RETRIES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_TIMEOUT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.HOSTNAME; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.METADATA_LIST; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PASSWORD; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PORT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_MODE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCHEMA_CHANGE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SERVER_TIME_ZONE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES_EXCLUDE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.USERNAME; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.getDebeziumProperties; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.PROPERTIES_PREFIX; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.getJdbcProperties; +import static org.apache.flink.util.Preconditions.checkState; + +/** A {@link Factory} to create {@link SqlServerDataSource}. */ +@Internal +public class SqlServerDataSourceFactory implements DataSourceFactory { + + private static final Logger LOG = LoggerFactory.getLogger(SqlServerDataSourceFactory.class); + + public static final String IDENTIFIER = "sqlserver"; + + @Override + public DataSource createDataSource(Context context) { + FactoryHelper.createFactoryHelper(this, context) + .validateExcept(PROPERTIES_PREFIX, DEBEZIUM_OPTIONS_PREFIX); + + final Configuration config = context.getFactoryConfiguration(); + String hostname = config.get(HOSTNAME); + int port = config.get(PORT); + String username = config.get(USERNAME); + String password = config.get(PASSWORD); + String chunkKeyColumn = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN); + String tables = config.get(TABLES); + ZoneId serverTimeZone = getServerTimeZone(config); + String tablesExclude = config.get(TABLES_EXCLUDE); + StartupOptions startupOptions = getStartupOptions(config); + + boolean includeSchemaChanges = config.get(SCHEMA_CHANGE_ENABLED); + + int fetchSize = config.get(SCAN_SNAPSHOT_FETCH_SIZE); + int splitSize = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE); + int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE); + + double distributionFactorUpper = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND); + double distributionFactorLower = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); + + boolean closeIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); + boolean isAssignUnboundedChunkFirst = + config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED); + + Duration connectTimeout = config.get(CONNECT_TIMEOUT); + int connectMaxRetries = config.get(CONNECT_MAX_RETRIES); + int connectionPoolSize = config.get(CONNECTION_POOL_SIZE); + boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); + + validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1); + validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1); + validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1); + validateIntegerOption(CONNECTION_POOL_SIZE, connectionPoolSize, 1); + validateIntegerOption(CONNECT_MAX_RETRIES, connectMaxRetries, 0); + validateDistributionFactorUpper(distributionFactorUpper); + validateDistributionFactorLower(distributionFactorLower); + + Map<String, String> configMap = config.toMap(); + mergeJdbcPropertiesIntoDebeziumProperties(configMap); + String databaseName = getValidateDatabaseName(tables).orElseThrow(); Review Comment: Two issues in database-name extraction/validation:\n1) If all entries are skipped (e.g., due to format issues), `getValidateDatabaseName(...)` can return `Optional.empty()`, and the caller uses `.orElseThrow()` without a message, resulting in a `NoSuchElementException` that doesn't explain the config problem.\n2) The length validation error message uses `TABLES.key()` (the literal key name) instead of the invalid value (e.g., `trimmedTableName` or the full `tables` string), making troubleshooting harder.\nRecommendation: treat any non-`db.schema.table` entry as a configuration error (fail fast with `ValidationException`/`IllegalArgumentException` containing the actual option value) and avoid returning empty when the option is required. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerPipelineITCaseTest.java: ########## @@ -0,0 +1,468 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.source; + +import org.apache.flink.api.common.eventtime.WatermarkStrategy; +import org.apache.flink.api.common.restartstrategy.RestartStrategies; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.data.binary.BinaryStringData; +import org.apache.flink.cdc.common.event.CreateTableEvent; +import org.apache.flink.cdc.common.event.DataChangeEvent; +import org.apache.flink.cdc.common.event.Event; +import org.apache.flink.cdc.common.event.SchemaChangeEvent; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.factories.Factory; +import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Schema; +import org.apache.flink.cdc.common.source.FlinkSourceProvider; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.common.types.DataTypes; +import org.apache.flink.cdc.common.types.RowType; +import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.cdc.connectors.sqlserver.SqlServerTestBase; +import org.apache.flink.cdc.connectors.sqlserver.factory.SqlServerDataSourceFactory; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfigFactory; +import org.apache.flink.cdc.runtime.typeutils.BinaryRecordDataGenerator; +import org.apache.flink.cdc.runtime.typeutils.EventTypeInfo; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.planner.factories.TestValuesTableFactory; +import org.apache.flink.util.CloseableIterator; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import java.sql.Connection; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.testcontainers.containers.MSSQLServerContainer.MS_SQL_SERVER_PORT; + +/** Integration tests for SQL Server pipeline source. */ +public class SqlServerPipelineITCaseTest extends SqlServerTestBase { + private static final String DATABASE_NAME = "inventory"; + + private static final StreamExecutionEnvironment env = + StreamExecutionEnvironment.getExecutionEnvironment(); + + @SuppressWarnings("deprecation") + @BeforeEach + public void before() { + TestValuesTableFactory.clearAllData(); + env.setParallelism(4); + env.enableCheckpointing(200); + env.setRestartStrategy(RestartStrategies.noRestart()); + initializeSqlServerTable(DATABASE_NAME); + } + + @Test + public void testInitialStartupMode() throws Exception { + SqlServerSourceConfigFactory configFactory = + (SqlServerSourceConfigFactory) + new SqlServerSourceConfigFactory() + .hostname(MSSQL_SERVER_CONTAINER.getHost()) + .port(MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT)) + .username(MSSQL_SERVER_CONTAINER.getUsername()) + .password(MSSQL_SERVER_CONTAINER.getPassword()) + .databaseList(DATABASE_NAME) + .tableList("dbo.products") + .startupOptions(StartupOptions.initial()) + .serverTimeZone("UTC"); + + FlinkSourceProvider sourceProvider = + (FlinkSourceProvider) + new SqlServerDataSource(configFactory).getEventSourceProvider(); + CloseableIterator<Event> events = + env.fromSource( + sourceProvider.getSource(), + WatermarkStrategy.noWatermarks(), + SqlServerDataSourceFactory.IDENTIFIER, + new EventTypeInfo()) + .executeAndCollect(); + + TableId tableId = TableId.tableId(DATABASE_NAME, "dbo", "products"); + CreateTableEvent createTableEvent = getProductsCreateTableEvent(tableId); + + // generate snapshot data + List<Event> expectedSnapshot = getSnapshotExpected(tableId); + + // In this configuration, several subtasks might emit their corresponding CreateTableEvent + // to downstream. Since it is not possible to predict how many CreateTableEvents should we + // expect, we simply filter them out from expected sets, and assert there's at least one. + List<Event> actual = fetchResultsExcept(events, expectedSnapshot.size(), createTableEvent); + assertThat(actual.subList(0, expectedSnapshot.size())) + .containsExactlyInAnyOrder(expectedSnapshot.toArray(new Event[0])); + } + + @ParameterizedTest(name = "unboundedChunkFirst: {0}") + @ValueSource(booleans = {true, false}) + public void testInitialStartupModeWithOpts(boolean unboundedChunkFirst) throws Exception { + Configuration sourceConfiguration = new Configuration(); + sourceConfiguration.set( + SqlServerDataSourceOptions.HOSTNAME, MSSQL_SERVER_CONTAINER.getHost()); + sourceConfiguration.set( + SqlServerDataSourceOptions.PORT, + MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT)); + sourceConfiguration.set( + SqlServerDataSourceOptions.USERNAME, MSSQL_SERVER_CONTAINER.getUsername()); + sourceConfiguration.set( + SqlServerDataSourceOptions.PASSWORD, MSSQL_SERVER_CONTAINER.getPassword()); + sourceConfiguration.set( + SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP, false); + sourceConfiguration.set(SqlServerDataSourceOptions.TABLES, DATABASE_NAME + ".dbo.products"); + sourceConfiguration.set(SqlServerDataSourceOptions.SERVER_TIME_ZONE, "UTC"); + sourceConfiguration.set( + SqlServerDataSourceOptions.METADATA_LIST, + "database_name,schema_name,table_name,op_ts"); + sourceConfiguration.set( + SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED, + unboundedChunkFirst); + + Factory.Context context = + new FactoryHelper.DefaultContext( + sourceConfiguration, new Configuration(), this.getClass().getClassLoader()); + FlinkSourceProvider sourceProvider = + (FlinkSourceProvider) + new SqlServerDataSourceFactory() + .createDataSource(context) + .getEventSourceProvider(); + CloseableIterator<Event> events = + env.fromSource( + sourceProvider.getSource(), + WatermarkStrategy.noWatermarks(), + SqlServerDataSourceFactory.IDENTIFIER, + new EventTypeInfo()) + .executeAndCollect(); + + TableId tableId = TableId.tableId(DATABASE_NAME, "dbo", "products"); + CreateTableEvent createTableEvent = getProductsCreateTableEvent(tableId); + + // generate snapshot data + Map<String, String> meta = new HashMap<>(); + meta.put("database_name", DATABASE_NAME); + meta.put("schema_name", "dbo"); + meta.put("table_name", "products"); + meta.put("op_ts", "0"); + + // generate snapshot data + List<Event> expectedSnapshot = + getSnapshotExpected(tableId).stream() + .map( + event -> { + DataChangeEvent dataChangeEvent = (DataChangeEvent) event; + return DataChangeEvent.insertEvent( + dataChangeEvent.tableId(), + dataChangeEvent.after(), + meta); + }) + .collect(Collectors.toList()); + + String startTime = String.valueOf(System.currentTimeMillis()); + Thread.sleep(1000); + + List<Event> expectedLog = new ArrayList<>(); + + try (Connection connection = getJdbcConnection(); + Statement statement = connection.createStatement()) { + statement.execute("USE " + DATABASE_NAME); + + RowType rowType = + RowType.of( + new DataType[] { + DataTypes.INT().notNull(), + DataTypes.VARCHAR(255).notNull(), + DataTypes.VARCHAR(512), + DataTypes.DOUBLE() + }, + new String[] {"id", "name", "description", "weight"}); + BinaryRecordDataGenerator generator = new BinaryRecordDataGenerator(rowType); + statement.execute( + "INSERT INTO dbo.products(name,description,weight) VALUES ('scooter','c-2',5.5)"); // 110 + expectedLog.add( + DataChangeEvent.insertEvent( + tableId, + generator.generate( + new Object[] { + 110, + BinaryStringData.fromString("scooter"), + BinaryStringData.fromString("c-2"), + 5.5d + }))); + statement.execute( + "INSERT INTO dbo.products(name,description,weight) VALUES ('football','c-11',6.6)"); // 111 + expectedLog.add( + DataChangeEvent.insertEvent( + tableId, + generator.generate( + new Object[] { + 111, + BinaryStringData.fromString("football"), + BinaryStringData.fromString("c-11"), + 6.6d + }))); + statement.execute("UPDATE dbo.products SET description='c-12' WHERE id=110"); + + expectedLog.add( + DataChangeEvent.updateEvent( + tableId, + generator.generate( + new Object[] { + 110, + BinaryStringData.fromString("scooter"), + BinaryStringData.fromString("c-2"), + 5.5d + }), + generator.generate( + new Object[] { + 110, + BinaryStringData.fromString("scooter"), + BinaryStringData.fromString("c-12"), + 5.5d + }))); + + statement.execute("DELETE FROM dbo.products WHERE id = 111"); + expectedLog.add( + DataChangeEvent.deleteEvent( + tableId, + generator.generate( + new Object[] { + 111, + BinaryStringData.fromString("football"), + BinaryStringData.fromString("c-11"), + 6.6d + }))); + } + + // In this configuration, several subtasks might emit their corresponding CreateTableEvent + // to downstream. Since it is not possible to predict how many CreateTableEvents should we + // expect, we simply filter them out from expected sets, and assert there's at least one. + int snapshotRecordsCount = expectedSnapshot.size(); + int logRecordsCount = expectedLog.size(); + + // Ditto, CreateTableEvent might be emitted in multiple partitions. + List<Event> actual = + fetchResultsExcept( + events, snapshotRecordsCount + logRecordsCount, createTableEvent); + + List<Event> actualSnapshotEvents = actual.subList(0, snapshotRecordsCount); + List<Event> actualLogEvents = actual.subList(snapshotRecordsCount, actual.size()); + + assertThat(actualSnapshotEvents).containsExactlyInAnyOrderElementsOf(expectedSnapshot); + assertThat(actualLogEvents).hasSize(logRecordsCount); + + for (int i = 0; i < logRecordsCount; i++) { + if (expectedLog.get(i) instanceof SchemaChangeEvent) { + assertThat(actualLogEvents.get(i)).isEqualTo(expectedLog.get(i)); + } else { + DataChangeEvent expectedEvent = (DataChangeEvent) expectedLog.get(i); + DataChangeEvent actualEvent = (DataChangeEvent) actualLogEvents.get(i); + assertThat(actualEvent.op()).isEqualTo(expectedEvent.op()); + assertThat(actualEvent.before()).isEqualTo(expectedEvent.before()); + assertThat(actualEvent.after()).isEqualTo(expectedEvent.after()); + assertThat(actualEvent.meta().get("database_name")).isEqualTo(DATABASE_NAME); + assertThat(actualEvent.meta().get("schema_name")).isEqualTo("dbo"); + assertThat(actualEvent.meta().get("table_name")).isEqualTo("products"); + assertThat(actualEvent.meta().get("op_ts")).isGreaterThanOrEqualTo(startTime); + } + } + } + + @Test + public void testSnapshotOnlyMode() throws Exception { + SqlServerSourceConfigFactory configFactory = + (SqlServerSourceConfigFactory) + new SqlServerSourceConfigFactory() + .hostname(MSSQL_SERVER_CONTAINER.getHost()) + .port(MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT)) + .username(MSSQL_SERVER_CONTAINER.getUsername()) + .password(MSSQL_SERVER_CONTAINER.getPassword()) + .databaseList(DATABASE_NAME) + .tableList("dbo.products") + .startupOptions(StartupOptions.snapshot()) + .skipSnapshotBackfill(false) + .serverTimeZone("UTC"); + + FlinkSourceProvider sourceProvider = + (FlinkSourceProvider) + new SqlServerDataSource(configFactory).getEventSourceProvider(); + CloseableIterator<Event> events = + env.fromSource( + sourceProvider.getSource(), + WatermarkStrategy.noWatermarks(), + SqlServerDataSourceFactory.IDENTIFIER, + new EventTypeInfo()) + .executeAndCollect(); + + TableId tableId = TableId.tableId(DATABASE_NAME, "dbo", "products"); + CreateTableEvent createTableEvent = getProductsCreateTableEvent(tableId); + + // generate snapshot data + List<Event> expectedSnapshot = getSnapshotExpected(tableId); + + // In this configuration, several subtasks might emit their corresponding CreateTableEvent + // to downstream. Since it is not possible to predict how many CreateTableEvents should we + // expect, we simply filter them out from expected sets, and assert there's at least one. + List<Event> actual = fetchResultsExcept(events, expectedSnapshot.size(), createTableEvent); + assertThat(actual.subList(0, expectedSnapshot.size())) + .containsExactlyInAnyOrder(expectedSnapshot.toArray(new Event[0])); + } + + private static <T> List<T> fetchResultsExcept(Iterator<T> iter, int size, T sideEvent) { + List<T> result = new ArrayList<>(size); + List<T> sideResults = new ArrayList<>(); + while (size > 0 && iter.hasNext()) { + T event = iter.next(); + if (sideEvent.getClass().isInstance(event)) { + sideResults.add(sideEvent); + } else { + result.add(event); + size--; + } + } + // Also ensure we've received at least one or many side events. + assertThat(sideResults).isNotEmpty(); + return result; + } Review Comment: When a side event is encountered, the code adds the `sideEvent` *parameter* rather than the actual `event` read from the iterator (`sideResults.add(sideEvent)`). This is logically incorrect and can hide issues if the test later inspects side events. Add the actual `event` instead. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerSchemaDataTypeInference.java: ########## @@ -0,0 +1,36 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.source; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.types.DataType; +import org.apache.flink.cdc.debezium.event.DebeziumSchemaDataTypeInference; + +import org.apache.kafka.connect.data.Schema; + +/** {@link DataType} inference for SQL Server debezium {@link Schema}. */ +@Internal +public class SqlServerSchemaDataTypeInference extends DebeziumSchemaDataTypeInference { + + private static final long serialVersionUID = 1L; + + // SQL Server doesn't have special geometry types like PostgreSQL, + // so we can use the default implementation from the parent class. + // If there are SQL Server specific types that need special handling, + // they can be added here by overriding the inferStruct method. Review Comment: The comment is misleading: SQL Server does have spatial types (`geometry`, `geography`) and other special types (e.g., `hierarchyid`). If the intent is 'no special handling needed currently', consider rephrasing to avoid stating that these types don't exist. ```suggestion // SQL Server has database-specific types, but no special handling is currently // needed here, so this class uses the default implementation from the parent class. // If SQL Server-specific types require special handling in the future, // it can be added here by overriding the inferStruct method. ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactory.java: ########## @@ -0,0 +1,442 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.factory; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.factories.DataSourceFactory; +import org.apache.flink.cdc.common.factories.Factory; +import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Selectors; +import org.apache.flink.cdc.common.source.DataSource; +import org.apache.flink.cdc.common.utils.StringUtils; +import org.apache.flink.cdc.connectors.base.options.SourceOptions; +import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSource; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfigFactory; +import org.apache.flink.cdc.connectors.sqlserver.table.SqlServerReadableMetadata; +import org.apache.flink.cdc.connectors.sqlserver.utils.SqlServerSchemaUtils; +import org.apache.flink.table.api.ValidationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Properties; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_META_GROUP_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECTION_POOL_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_MAX_RETRIES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_TIMEOUT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.HOSTNAME; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.METADATA_LIST; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PASSWORD; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PORT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_MODE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCHEMA_CHANGE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SERVER_TIME_ZONE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES_EXCLUDE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.USERNAME; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.getDebeziumProperties; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.PROPERTIES_PREFIX; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.getJdbcProperties; +import static org.apache.flink.util.Preconditions.checkState; + +/** A {@link Factory} to create {@link SqlServerDataSource}. */ +@Internal +public class SqlServerDataSourceFactory implements DataSourceFactory { + + private static final Logger LOG = LoggerFactory.getLogger(SqlServerDataSourceFactory.class); + + public static final String IDENTIFIER = "sqlserver"; + + @Override + public DataSource createDataSource(Context context) { + FactoryHelper.createFactoryHelper(this, context) + .validateExcept(PROPERTIES_PREFIX, DEBEZIUM_OPTIONS_PREFIX); + + final Configuration config = context.getFactoryConfiguration(); + String hostname = config.get(HOSTNAME); + int port = config.get(PORT); + String username = config.get(USERNAME); + String password = config.get(PASSWORD); + String chunkKeyColumn = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN); + String tables = config.get(TABLES); + ZoneId serverTimeZone = getServerTimeZone(config); + String tablesExclude = config.get(TABLES_EXCLUDE); + StartupOptions startupOptions = getStartupOptions(config); + + boolean includeSchemaChanges = config.get(SCHEMA_CHANGE_ENABLED); + + int fetchSize = config.get(SCAN_SNAPSHOT_FETCH_SIZE); + int splitSize = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE); + int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE); + + double distributionFactorUpper = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND); + double distributionFactorLower = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); + + boolean closeIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); + boolean isAssignUnboundedChunkFirst = + config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED); + + Duration connectTimeout = config.get(CONNECT_TIMEOUT); + int connectMaxRetries = config.get(CONNECT_MAX_RETRIES); + int connectionPoolSize = config.get(CONNECTION_POOL_SIZE); + boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); + + validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1); + validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1); + validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1); + validateIntegerOption(CONNECTION_POOL_SIZE, connectionPoolSize, 1); + validateIntegerOption(CONNECT_MAX_RETRIES, connectMaxRetries, 0); + validateDistributionFactorUpper(distributionFactorUpper); + validateDistributionFactorLower(distributionFactorLower); + + Map<String, String> configMap = config.toMap(); + mergeJdbcPropertiesIntoDebeziumProperties(configMap); + String databaseName = getValidateDatabaseName(tables).orElseThrow(); + + SqlServerSourceConfigFactory configFactory = new SqlServerSourceConfigFactory(); + configFactory + .hostname(hostname) + .port(port) + .databaseList(databaseName) + .tableList(".*\\..*") + .username(username) + .password(password) + .serverTimeZone(serverTimeZone.getId()) + .debeziumProperties(getDebeziumProperties(configMap)) + .splitSize(splitSize) + .splitMetaGroupSize(splitMetaGroupSize) + .distributionFactorUpper(distributionFactorUpper) + .distributionFactorLower(distributionFactorLower) + .fetchSize(fetchSize) + .connectTimeout(connectTimeout) + .connectMaxRetries(connectMaxRetries) + .connectionPoolSize(connectionPoolSize) + .includeSchemaChanges(includeSchemaChanges) + .startupOptions(startupOptions) + .chunkKeyColumn(chunkKeyColumn) + .closeIdleReaders(closeIdleReaders) + .skipSnapshotBackfill(skipSnapshotBackfill) + .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled) + .assignUnboundedChunkFirst(isAssignUnboundedChunkFirst); + + SqlServerSourceConfig sourceConfig = configFactory.create(0); + SqlServerSchemaUtils.validateSqlServerAgentRunning(sourceConfig); + + List<TableId> tableIds = SqlServerSchemaUtils.listTables(sourceConfig, null, null); + + Selectors selectors = new Selectors.SelectorsBuilder().includeTables(tables).build(); + List<String> capturedTables = getTableList(tableIds, selectors); + if (capturedTables.isEmpty()) { + throw new IllegalArgumentException( + "Cannot find any table by the option 'tables' = " + tables); + } + if (tablesExclude != null) { + Selectors selectExclude = + new Selectors.SelectorsBuilder().includeTables(tablesExclude).build(); + List<String> excludeTables = getTableList(tableIds, selectExclude); + if (!excludeTables.isEmpty()) { + capturedTables.removeAll(excludeTables); + } + if (capturedTables.isEmpty()) { + throw new IllegalArgumentException( + "Cannot find any table with by the option 'tables.exclude' = " + + tablesExclude); + } + } + configFactory.tableList(capturedTables.toArray(new String[0])); + + String metadataList = config.get(METADATA_LIST); + List<SqlServerReadableMetadata> readableMetadataList = listReadableMetadata(metadataList); + + return new SqlServerDataSource(configFactory, readableMetadataList); + } + + private void mergeJdbcPropertiesIntoDebeziumProperties(Map<String, String> configMap) { + Properties jdbcProperties = getJdbcProperties(configMap); + for (String propertyName : jdbcProperties.stringPropertyNames()) { + configMap.putIfAbsent( + DEBEZIUM_OPTIONS_PREFIX + "database." + propertyName, + jdbcProperties.getProperty(propertyName)); + } + } + + private List<SqlServerReadableMetadata> listReadableMetadata(String metadataList) { + if (StringUtils.isNullOrWhitespaceOnly(metadataList)) { + return new ArrayList<>(); + } + Set<String> readableMetadataSet = + Arrays.stream(metadataList.split(",")) + .map(String::trim) + .collect(Collectors.toSet()); + List<SqlServerReadableMetadata> foundMetadata = new ArrayList<>(); + for (SqlServerReadableMetadata metadata : SqlServerReadableMetadata.values()) { + if (readableMetadataSet.contains(metadata.getKey())) { + foundMetadata.add(metadata); + readableMetadataSet.remove(metadata.getKey()); + } + } + if (readableMetadataSet.isEmpty()) { + return foundMetadata; + } + throw new IllegalArgumentException( + String.format( + "[%s] cannot be found in SQL Server metadata.", + String.join(", ", readableMetadataSet))); + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(HOSTNAME); + options.add(USERNAME); + options.add(PASSWORD); + options.add(TABLES); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PORT); + options.add(TABLES_EXCLUDE); + options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE); + options.add(SCAN_SNAPSHOT_FETCH_SIZE); + options.add(SCHEMA_CHANGE_ENABLED); + options.add(SCAN_STARTUP_MODE); + options.add(SCAN_STARTUP_TIMESTAMP_MILLIS); + options.add(SERVER_TIME_ZONE); + options.add(CONNECT_TIMEOUT); + options.add(CONNECT_MAX_RETRIES); + options.add(CONNECTION_POOL_SIZE); + options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); + options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); + options.add(CHUNK_META_GROUP_SIZE); + options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND); + options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); + options.add(METADATA_LIST); + options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED); + return options; + } + + @Override + public String identifier() { + return IDENTIFIER; + } + + private static List<String> getTableList( + @Nullable List<TableId> tableIdList, Selectors selectors) { + return Objects.requireNonNullElse(tableIdList, Collections.<TableId>emptyList()).stream() + .filter(selectors::isMatch) + // SQL Server tableList format: schemaName.tableName (without database prefix) + // See SqlServerSourceBuilder: "Each identifier is of the form + // <schemaName>.<tableName>" + .map(tableId -> tableId.getSchemaName() + "." + tableId.getTableName()) + .collect(Collectors.toList()); + } + + /** Checks the value of given integer option is valid. */ + private void validateIntegerOption( + ConfigOption<Integer> option, int optionValue, int exclusiveMin) { + checkState( + optionValue > exclusiveMin, + String.format( + "The value of option '%s' must larger than %d, but is %d", + option.key(), exclusiveMin, optionValue)); + } + + private static final String SCAN_STARTUP_MODE_VALUE_INITIAL = "initial"; + + private static final String SCAN_STARTUP_MODE_VALUE_SNAPSHOT = "snapshot"; + private static final String SCAN_STARTUP_MODE_VALUE_LATEST = "latest-offset"; + private static final String SCAN_STARTUP_MODE_VALUE_TIMESTAMP = "timestamp"; + + private static StartupOptions getStartupOptions(Configuration config) { + String modeString = config.get(SCAN_STARTUP_MODE); + + switch (modeString.toLowerCase()) { + case SCAN_STARTUP_MODE_VALUE_INITIAL: + return StartupOptions.initial(); + case SCAN_STARTUP_MODE_VALUE_SNAPSHOT: + return StartupOptions.snapshot(); + case SCAN_STARTUP_MODE_VALUE_LATEST: + return StartupOptions.latest(); + + case SCAN_STARTUP_MODE_VALUE_TIMESTAMP: + Long startupTimestampMillis = config.get(SCAN_STARTUP_TIMESTAMP_MILLIS); + if (startupTimestampMillis == null) { + throw new ValidationException( + String.format( + "To use timestamp startup mode, the startup timestamp millis '%s' must be set.", + SCAN_STARTUP_TIMESTAMP_MILLIS.key())); + } + return StartupOptions.timestamp(startupTimestampMillis); + + default: + throw new ValidationException( + String.format( + "Invalid value for option '%s'. Supported values are [%s, %s, %s, %s], but was: %s", + SourceOptions.SCAN_STARTUP_MODE.key(), + SCAN_STARTUP_MODE_VALUE_INITIAL, + SCAN_STARTUP_MODE_VALUE_SNAPSHOT, + SCAN_STARTUP_MODE_VALUE_LATEST, + SCAN_STARTUP_MODE_VALUE_TIMESTAMP, + modeString)); + } + } + + /** Checks the value of given evenly distribution factor upper bound is valid. */ + private void validateDistributionFactorUpper(double distributionFactorUpper) { + checkState( + doubleCompare(distributionFactorUpper, 1.0d) >= 0, + String.format( + "The value of option '%s' must larger than or equals %s, but is %s", + SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.key(), + 1.0d, + distributionFactorUpper)); + } + + /** Checks the value of given evenly distribution factor lower bound is valid. */ + private void validateDistributionFactorLower(double distributionFactorLower) { + checkState( + doubleCompare(distributionFactorLower, 0.0d) >= 0 + && doubleCompare(distributionFactorLower, 1.0d) <= 0, + String.format( + "The value of option '%s' must between %s and %s inclusively, but is %s", + SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.key(), + 0.0d, + 1.0d, + distributionFactorLower)); + } + + /** + * Get the database name from tables configuration. + * + * @param tables Table name list, format is "db.schema.table,db.schema.table,..." Each table + * name consists of three parts separated by ".", which are database name, schema name, and + * table name. + * @return Database name if found, otherwise returns Optional.empty() + * @throws IllegalArgumentException If the input parameter is null or does not match the + * expected format, or if database names are inconsistent. + */ + private Optional<String> getValidateDatabaseName(String tables) { + // Input validation + if (tables == null || tables.trim().isEmpty()) { + throw new IllegalArgumentException("Parameter tables cannot be null or empty"); + } Review Comment: Two issues in database-name extraction/validation:\n1) If all entries are skipped (e.g., due to format issues), `getValidateDatabaseName(...)` can return `Optional.empty()`, and the caller uses `.orElseThrow()` without a message, resulting in a `NoSuchElementException` that doesn't explain the config problem.\n2) The length validation error message uses `TABLES.key()` (the literal key name) instead of the invalid value (e.g., `trimmedTableName` or the full `tables` string), making troubleshooting harder.\nRecommendation: treat any non-`db.schema.table` entry as a configuration error (fail fast with `ValidationException`/`IllegalArgumentException` containing the actual option value) and avoid returning empty when the option is required. ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerDataSourceOptions.java: ########## @@ -0,0 +1,258 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.source; + +import org.apache.flink.cdc.common.annotation.Experimental; +import org.apache.flink.cdc.common.annotation.PublicEvolving; +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.ConfigOptions; + +import java.time.Duration; + +/** Configurations for {@link SqlServerDataSource}. */ +@PublicEvolving +public class SqlServerDataSourceOptions { + + public static final ConfigOption<String> HOSTNAME = + ConfigOptions.key("hostname") + .stringType() + .noDefaultValue() + .withDescription("IP address or hostname of the SQL Server database server."); + + public static final ConfigOption<Integer> PORT = + ConfigOptions.key("port") + .intType() + .defaultValue(1433) + .withDescription("Integer port number of the SQL Server database server."); + + public static final ConfigOption<String> USERNAME = + ConfigOptions.key("username") + .stringType() + .noDefaultValue() + .withDescription( + "Name of the SQL Server database to use when connecting to the SQL Server database server."); Review Comment: The `username` option description is incorrect: it describes a database name rather than the user name. Update the description to avoid confusing users configuring the connector. ```suggestion "Name of the SQL Server user to use when connecting to the SQL Server database server."); ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/testutils/SqlServerSourceTestUtils.java: ########## @@ -0,0 +1,75 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.testutils; + +import org.apache.flink.api.java.tuple.Tuple2; +import org.apache.flink.cdc.common.event.CreateTableEvent; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Supplier; + +/** Test utilities for SQL Server event source. */ +public class SqlServerSourceTestUtils { + + public static <T> List<T> fetchResults(Iterator<T> iter, int size) { + List<T> result = new ArrayList<>(size); + while (size > 0 && iter.hasNext()) { + T event = iter.next(); + result.add(event); + size--; + } + return result; + } + + public static <T> Tuple2<List<T>, List<CreateTableEvent>> fetchResultsAndCreateTableEvent( + Iterator<T> iter, int size) { + List<T> result = new ArrayList<>(size); + List<CreateTableEvent> createTableEvents = new ArrayList<>(); + while (size > 0 && iter.hasNext()) { + T event = iter.next(); + if (event instanceof CreateTableEvent) { + createTableEvents.add((CreateTableEvent) event); + } else { + result.add(event); + size--; + } + } + return Tuple2.of(result, createTableEvents); + } + + public static void loopCheck( + Supplier<Boolean> runnable, String description, Duration timeout, Duration interval) + throws Exception { + long deadline = System.currentTimeMillis() + timeout.toMillis(); + while (System.currentTimeMillis() < deadline) { + if (runnable.get()) { + return; + } + TimeUnit.MILLISECONDS.sleep(interval.toMillis()); + } + throw new TimeoutException( + "Ran out of time when waiting for " + description + " to success."); Review Comment: Fix grammar in the timeout message: replace 'to success' with 'to succeed'. ```suggestion "Ran out of time when waiting for " + description + " to succeed."); ``` ########## flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactory.java: ########## @@ -0,0 +1,442 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.cdc.connectors.sqlserver.factory; + +import org.apache.flink.cdc.common.annotation.Internal; +import org.apache.flink.cdc.common.configuration.ConfigOption; +import org.apache.flink.cdc.common.configuration.Configuration; +import org.apache.flink.cdc.common.event.TableId; +import org.apache.flink.cdc.common.factories.DataSourceFactory; +import org.apache.flink.cdc.common.factories.Factory; +import org.apache.flink.cdc.common.factories.FactoryHelper; +import org.apache.flink.cdc.common.schema.Selectors; +import org.apache.flink.cdc.common.source.DataSource; +import org.apache.flink.cdc.common.utils.StringUtils; +import org.apache.flink.cdc.connectors.base.options.SourceOptions; +import org.apache.flink.cdc.connectors.base.options.StartupOptions; +import org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSource; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig; +import org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfigFactory; +import org.apache.flink.cdc.connectors.sqlserver.table.SqlServerReadableMetadata; +import org.apache.flink.cdc.connectors.sqlserver.utils.SqlServerSchemaUtils; +import org.apache.flink.table.api.ValidationException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.annotation.Nullable; + +import java.time.Duration; +import java.time.ZoneId; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.Properties; +import java.util.Set; +import java.util.stream.Collectors; + +import static org.apache.flink.cdc.connectors.base.utils.ObjectUtils.doubleCompare; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CHUNK_META_GROUP_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECTION_POOL_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_MAX_RETRIES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.CONNECT_TIMEOUT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.HOSTNAME; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.METADATA_LIST; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PASSWORD; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.PORT; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_NEWLY_ADDED_TABLE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_MODE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCAN_STARTUP_TIMESTAMP_MILLIS; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SCHEMA_CHANGE_ENABLED; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SERVER_TIME_ZONE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.TABLES_EXCLUDE; +import static org.apache.flink.cdc.connectors.sqlserver.source.SqlServerDataSourceOptions.USERNAME; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.DEBEZIUM_OPTIONS_PREFIX; +import static org.apache.flink.cdc.debezium.table.DebeziumOptions.getDebeziumProperties; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.PROPERTIES_PREFIX; +import static org.apache.flink.cdc.debezium.utils.JdbcUrlUtils.getJdbcProperties; +import static org.apache.flink.util.Preconditions.checkState; + +/** A {@link Factory} to create {@link SqlServerDataSource}. */ +@Internal +public class SqlServerDataSourceFactory implements DataSourceFactory { + + private static final Logger LOG = LoggerFactory.getLogger(SqlServerDataSourceFactory.class); + + public static final String IDENTIFIER = "sqlserver"; + + @Override + public DataSource createDataSource(Context context) { + FactoryHelper.createFactoryHelper(this, context) + .validateExcept(PROPERTIES_PREFIX, DEBEZIUM_OPTIONS_PREFIX); + + final Configuration config = context.getFactoryConfiguration(); + String hostname = config.get(HOSTNAME); + int port = config.get(PORT); + String username = config.get(USERNAME); + String password = config.get(PASSWORD); + String chunkKeyColumn = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_KEY_COLUMN); + String tables = config.get(TABLES); + ZoneId serverTimeZone = getServerTimeZone(config); + String tablesExclude = config.get(TABLES_EXCLUDE); + StartupOptions startupOptions = getStartupOptions(config); + + boolean includeSchemaChanges = config.get(SCHEMA_CHANGE_ENABLED); + + int fetchSize = config.get(SCAN_SNAPSHOT_FETCH_SIZE); + int splitSize = config.get(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE); + int splitMetaGroupSize = config.get(CHUNK_META_GROUP_SIZE); + + double distributionFactorUpper = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND); + double distributionFactorLower = config.get(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); + + boolean closeIdleReaders = config.get(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); + boolean isAssignUnboundedChunkFirst = + config.get(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + boolean scanNewlyAddedTableEnabled = config.get(SCAN_NEWLY_ADDED_TABLE_ENABLED); + + Duration connectTimeout = config.get(CONNECT_TIMEOUT); + int connectMaxRetries = config.get(CONNECT_MAX_RETRIES); + int connectionPoolSize = config.get(CONNECTION_POOL_SIZE); + boolean skipSnapshotBackfill = config.get(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); + + validateIntegerOption(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE, splitSize, 1); + validateIntegerOption(CHUNK_META_GROUP_SIZE, splitMetaGroupSize, 1); + validateIntegerOption(SCAN_SNAPSHOT_FETCH_SIZE, fetchSize, 1); + validateIntegerOption(CONNECTION_POOL_SIZE, connectionPoolSize, 1); + validateIntegerOption(CONNECT_MAX_RETRIES, connectMaxRetries, 0); + validateDistributionFactorUpper(distributionFactorUpper); + validateDistributionFactorLower(distributionFactorLower); + + Map<String, String> configMap = config.toMap(); + mergeJdbcPropertiesIntoDebeziumProperties(configMap); + String databaseName = getValidateDatabaseName(tables).orElseThrow(); + + SqlServerSourceConfigFactory configFactory = new SqlServerSourceConfigFactory(); + configFactory + .hostname(hostname) + .port(port) + .databaseList(databaseName) + .tableList(".*\\..*") + .username(username) + .password(password) + .serverTimeZone(serverTimeZone.getId()) + .debeziumProperties(getDebeziumProperties(configMap)) + .splitSize(splitSize) + .splitMetaGroupSize(splitMetaGroupSize) + .distributionFactorUpper(distributionFactorUpper) + .distributionFactorLower(distributionFactorLower) + .fetchSize(fetchSize) + .connectTimeout(connectTimeout) + .connectMaxRetries(connectMaxRetries) + .connectionPoolSize(connectionPoolSize) + .includeSchemaChanges(includeSchemaChanges) + .startupOptions(startupOptions) + .chunkKeyColumn(chunkKeyColumn) + .closeIdleReaders(closeIdleReaders) + .skipSnapshotBackfill(skipSnapshotBackfill) + .scanNewlyAddedTableEnabled(scanNewlyAddedTableEnabled) + .assignUnboundedChunkFirst(isAssignUnboundedChunkFirst); + + SqlServerSourceConfig sourceConfig = configFactory.create(0); + SqlServerSchemaUtils.validateSqlServerAgentRunning(sourceConfig); + + List<TableId> tableIds = SqlServerSchemaUtils.listTables(sourceConfig, null, null); + + Selectors selectors = new Selectors.SelectorsBuilder().includeTables(tables).build(); + List<String> capturedTables = getTableList(tableIds, selectors); + if (capturedTables.isEmpty()) { + throw new IllegalArgumentException( + "Cannot find any table by the option 'tables' = " + tables); + } + if (tablesExclude != null) { + Selectors selectExclude = + new Selectors.SelectorsBuilder().includeTables(tablesExclude).build(); + List<String> excludeTables = getTableList(tableIds, selectExclude); + if (!excludeTables.isEmpty()) { + capturedTables.removeAll(excludeTables); + } + if (capturedTables.isEmpty()) { + throw new IllegalArgumentException( + "Cannot find any table with by the option 'tables.exclude' = " + + tablesExclude); + } + } + configFactory.tableList(capturedTables.toArray(new String[0])); + + String metadataList = config.get(METADATA_LIST); + List<SqlServerReadableMetadata> readableMetadataList = listReadableMetadata(metadataList); + + return new SqlServerDataSource(configFactory, readableMetadataList); + } + + private void mergeJdbcPropertiesIntoDebeziumProperties(Map<String, String> configMap) { + Properties jdbcProperties = getJdbcProperties(configMap); + for (String propertyName : jdbcProperties.stringPropertyNames()) { + configMap.putIfAbsent( + DEBEZIUM_OPTIONS_PREFIX + "database." + propertyName, + jdbcProperties.getProperty(propertyName)); + } + } + + private List<SqlServerReadableMetadata> listReadableMetadata(String metadataList) { + if (StringUtils.isNullOrWhitespaceOnly(metadataList)) { + return new ArrayList<>(); + } + Set<String> readableMetadataSet = + Arrays.stream(metadataList.split(",")) + .map(String::trim) + .collect(Collectors.toSet()); + List<SqlServerReadableMetadata> foundMetadata = new ArrayList<>(); + for (SqlServerReadableMetadata metadata : SqlServerReadableMetadata.values()) { + if (readableMetadataSet.contains(metadata.getKey())) { + foundMetadata.add(metadata); + readableMetadataSet.remove(metadata.getKey()); + } + } + if (readableMetadataSet.isEmpty()) { + return foundMetadata; + } + throw new IllegalArgumentException( + String.format( + "[%s] cannot be found in SQL Server metadata.", + String.join(", ", readableMetadataSet))); + } + + @Override + public Set<ConfigOption<?>> requiredOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(HOSTNAME); + options.add(USERNAME); + options.add(PASSWORD); + options.add(TABLES); + return options; + } + + @Override + public Set<ConfigOption<?>> optionalOptions() { + Set<ConfigOption<?>> options = new HashSet<>(); + options.add(PORT); + options.add(TABLES_EXCLUDE); + options.add(SCAN_INCREMENTAL_SNAPSHOT_CHUNK_SIZE); + options.add(SCAN_SNAPSHOT_FETCH_SIZE); + options.add(SCHEMA_CHANGE_ENABLED); + options.add(SCAN_STARTUP_MODE); + options.add(SCAN_STARTUP_TIMESTAMP_MILLIS); + options.add(SERVER_TIME_ZONE); + options.add(CONNECT_TIMEOUT); + options.add(CONNECT_MAX_RETRIES); + options.add(CONNECTION_POOL_SIZE); + options.add(SCAN_INCREMENTAL_CLOSE_IDLE_READER_ENABLED); + options.add(SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP); + options.add(CHUNK_META_GROUP_SIZE); + options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND); + options.add(CHUNK_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND); + options.add(METADATA_LIST); + options.add(SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED); + options.add(SCAN_NEWLY_ADDED_TABLE_ENABLED); + return options; + } + + @Override + public String identifier() { + return IDENTIFIER; + } + + private static List<String> getTableList( + @Nullable List<TableId> tableIdList, Selectors selectors) { + return Objects.requireNonNullElse(tableIdList, Collections.<TableId>emptyList()).stream() + .filter(selectors::isMatch) + // SQL Server tableList format: schemaName.tableName (without database prefix) + // See SqlServerSourceBuilder: "Each identifier is of the form + // <schemaName>.<tableName>" + .map(tableId -> tableId.getSchemaName() + "." + tableId.getTableName()) + .collect(Collectors.toList()); + } + + /** Checks the value of given integer option is valid. */ + private void validateIntegerOption( + ConfigOption<Integer> option, int optionValue, int exclusiveMin) { + checkState( + optionValue > exclusiveMin, + String.format( + "The value of option '%s' must larger than %d, but is %d", + option.key(), exclusiveMin, optionValue)); + } + + private static final String SCAN_STARTUP_MODE_VALUE_INITIAL = "initial"; + + private static final String SCAN_STARTUP_MODE_VALUE_SNAPSHOT = "snapshot"; + private static final String SCAN_STARTUP_MODE_VALUE_LATEST = "latest-offset"; + private static final String SCAN_STARTUP_MODE_VALUE_TIMESTAMP = "timestamp"; + + private static StartupOptions getStartupOptions(Configuration config) { + String modeString = config.get(SCAN_STARTUP_MODE); + + switch (modeString.toLowerCase()) { + case SCAN_STARTUP_MODE_VALUE_INITIAL: + return StartupOptions.initial(); + case SCAN_STARTUP_MODE_VALUE_SNAPSHOT: + return StartupOptions.snapshot(); + case SCAN_STARTUP_MODE_VALUE_LATEST: + return StartupOptions.latest(); + + case SCAN_STARTUP_MODE_VALUE_TIMESTAMP: + Long startupTimestampMillis = config.get(SCAN_STARTUP_TIMESTAMP_MILLIS); + if (startupTimestampMillis == null) { + throw new ValidationException( + String.format( + "To use timestamp startup mode, the startup timestamp millis '%s' must be set.", + SCAN_STARTUP_TIMESTAMP_MILLIS.key())); + } + return StartupOptions.timestamp(startupTimestampMillis); + + default: + throw new ValidationException( + String.format( + "Invalid value for option '%s'. Supported values are [%s, %s, %s, %s], but was: %s", + SourceOptions.SCAN_STARTUP_MODE.key(), + SCAN_STARTUP_MODE_VALUE_INITIAL, + SCAN_STARTUP_MODE_VALUE_SNAPSHOT, + SCAN_STARTUP_MODE_VALUE_LATEST, + SCAN_STARTUP_MODE_VALUE_TIMESTAMP, + modeString)); + } + } + + /** Checks the value of given evenly distribution factor upper bound is valid. */ + private void validateDistributionFactorUpper(double distributionFactorUpper) { + checkState( + doubleCompare(distributionFactorUpper, 1.0d) >= 0, + String.format( + "The value of option '%s' must larger than or equals %s, but is %s", + SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_UPPER_BOUND.key(), + 1.0d, + distributionFactorUpper)); + } + + /** Checks the value of given evenly distribution factor lower bound is valid. */ + private void validateDistributionFactorLower(double distributionFactorLower) { + checkState( + doubleCompare(distributionFactorLower, 0.0d) >= 0 + && doubleCompare(distributionFactorLower, 1.0d) <= 0, + String.format( + "The value of option '%s' must between %s and %s inclusively, but is %s", + SPLIT_KEY_EVEN_DISTRIBUTION_FACTOR_LOWER_BOUND.key(), + 0.0d, + 1.0d, + distributionFactorLower)); + } + + /** + * Get the database name from tables configuration. + * + * @param tables Table name list, format is "db.schema.table,db.schema.table,..." Each table + * name consists of three parts separated by ".", which are database name, schema name, and + * table name. + * @return Database name if found, otherwise returns Optional.empty() + * @throws IllegalArgumentException If the input parameter is null or does not match the + * expected format, or if database names are inconsistent. + */ + private Optional<String> getValidateDatabaseName(String tables) { + // Input validation + if (tables == null || tables.trim().isEmpty()) { + throw new IllegalArgumentException("Parameter tables cannot be null or empty"); + } + + // Split table name list + String[] tableNames = tables.split(","); + String dbName = null; + + for (String tableName : tableNames) { + // Trim whitespace and split table name + String trimmedTableName = tableName.trim(); + if (!trimmedTableName.contains(".")) { + continue; // Skip table names that do not match the expected format + } + + String[] tableNameParts = + trimmedTableName.split( + "(?<!\\\\)\\.", -1); // Use -1 to avoid ignoring trailing empty elements + + checkState( + tableNameParts.length == 3, + String.format( + "Tables format must db.schema.table, can not 'tables' = %s", + TABLES.key())); Review Comment: Two issues in database-name extraction/validation:\n1) If all entries are skipped (e.g., due to format issues), `getValidateDatabaseName(...)` can return `Optional.empty()`, and the caller uses `.orElseThrow()` without a message, resulting in a `NoSuchElementException` that doesn't explain the config problem.\n2) The length validation error message uses `TABLES.key()` (the literal key name) instead of the invalid value (e.g., `trimmedTableName` or the full `tables` string), making troubleshooting harder.\nRecommendation: treat any non-`db.schema.table` entry as a configuration error (fail fast with `ValidationException`/`IllegalArgumentException` containing the actual option value) and avoid returning empty when the option is required. -- 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]
