This is an automated email from the ASF dual-hosted git repository.
leonardBang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new b5152df63 [minor][pipeline-connector/sqlserver] Align MySQL's schema
restore process instead of per-record JDBC
b5152df63 is described below
commit b5152df634bc79fed05125ee748016bb48e7af12
Author: Leonard Xu <[email protected]>
AuthorDate: Thu Jul 2 16:33:56 2026 +0800
[minor][pipeline-connector/sqlserver] Align MySQL's schema restore process
instead of per-record JDBC
This closes #4446.
Co-authored-by: Claude Opus 4.7 <[email protected]>
---
.../reader/SqlServerPipelineRecordEmitter.java | 59 +--
.../SqlServerPipelineSavepointRestoreITCase.java | 428 +++++++++++++++++++++
.../reader/SqlServerPipelineRecordEmitterTest.java | 147 +++++++
.../source/reader/IncrementalSourceReader.java | 3 +
.../source/reader/IncrementalSourceReaderTest.java | 172 +++++++++
5 files changed, 784 insertions(+), 25 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitter.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitter.java
index 42bc908b6..cc789ed40 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitter.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitter.java
@@ -23,6 +23,8 @@ import org.apache.flink.cdc.common.event.TableId;
import org.apache.flink.cdc.common.schema.Schema;
import org.apache.flink.cdc.connectors.base.options.StartupOptions;
import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitState;
import org.apache.flink.cdc.connectors.base.source.metrics.SourceReaderMetrics;
import
org.apache.flink.cdc.connectors.base.source.reader.IncrementalSourceRecordEmitter;
@@ -39,6 +41,7 @@ import
io.debezium.relational.history.TableChanges.TableChange;
import org.apache.kafka.connect.source.SourceRecord;
import java.sql.SQLException;
+import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
@@ -83,7 +86,6 @@ public class SqlServerPipelineRecordEmitter<T> extends
IncrementalSourceRecordEm
((DebeziumEventDeserializationSchema)
debeziumDeserializationSchema)
.getCreateTableEventCache();
this.isBounded =
StartupOptions.snapshot().equals(sourceConfig.getStartupOptions());
- generateCreateTableEvents();
}
@Override
@@ -91,8 +93,7 @@ public class SqlServerPipelineRecordEmitter<T> extends
IncrementalSourceRecordEm
SourceRecord element, SourceOutput<T> output, SourceSplitState
splitState)
throws Exception {
if (isSchemaChangeEvent(element) && splitState.isStreamSplitState()) {
- restoreCreateTableEventsFromSplitSchemas(
- splitState.asStreamSplitState().getTableSchemas());
+
cacheCreateTableEventsFromSchemas(splitState.asStreamSplitState().getTableSchemas());
}
if (shouldEmitAllCreateTableEventsInSnapshotMode && isBounded) {
@@ -104,15 +105,25 @@ public class SqlServerPipelineRecordEmitter<T> extends
IncrementalSourceRecordEm
// to downstream to avoid checkpoint timeout.
io.debezium.relational.TableId tableId =
splitState.asSnapshotSplitState().toSourceSplit().getTableId();
- emitCreateTableEventIfNeeded(tableId, output);
+ emitCreateTableEventIfNeeded(tableId, output, splitState);
} else if (isDataChangeRecord(element)) {
// Handle data change events, schema change events are handled
downstream directly
io.debezium.relational.TableId tableId = getTableId(element);
- emitCreateTableEventIfNeeded(tableId, output);
+ emitCreateTableEventIfNeeded(tableId, output, splitState);
}
super.processElement(element, output, splitState);
}
+ @Override
+ public void applySplit(SourceSplitBase split) {
+ if (isBounded && createTableEventCache.isEmpty() && split instanceof
SnapshotSplit) {
+ // TableSchemas in SnapshotSplit only contains one table.
+ createTableEventCache.putAll(generateCreateTableEvents());
+ } else {
+ cacheCreateTableEventsFromSchemas(split.getTableSchemas());
+ }
+ }
+
@SuppressWarnings("unchecked")
private void emitAllCreateTableEvents(SourceOutput<T> output) {
createTableEventCache.forEach(
@@ -124,33 +135,28 @@ public class SqlServerPipelineRecordEmitter<T> extends
IncrementalSourceRecordEm
@SuppressWarnings("unchecked")
private void emitCreateTableEventIfNeeded(
- io.debezium.relational.TableId tableId, SourceOutput<T> output) {
+ io.debezium.relational.TableId tableId,
+ SourceOutput<T> output,
+ SourceSplitState splitState) {
if (alreadySendCreateTableTables.contains(tableId)) {
return;
}
+
cacheCreateTableEventsFromSchemas(splitState.toSourceSplit().getTableSchemas());
CreateTableEvent createTableEvent = createTableEventCache.get(tableId);
- if (createTableEvent != null) {
- output.collect((T) createTableEvent);
- } else {
- // Table not in cache, fetch schema from database
- try (SqlServerConnection jdbc =
-
createSqlServerConnection(sourceConfig.getDbzConnectorConfig())) {
- createTableEvent = buildCreateTableEvent(jdbc, tableId);
- output.collect((T) createTableEvent);
- createTableEventCache.put(tableId, createTableEvent);
- } catch (SQLException e) {
- throw new RuntimeException("Failed to get table schema for " +
tableId, e);
- }
+ if (createTableEvent == null) {
+ throw new IllegalStateException(
+ "Missing CreateTableEvent for table "
+ + tableId
+ + ". Table schema should have been restored before
processing records.");
}
+ output.collect((T) createTableEvent);
alreadySendCreateTableTables.add(tableId);
}
- private void restoreCreateTableEventsFromSplitSchemas(
+ private void cacheCreateTableEventsFromSchemas(
Map<io.debezium.relational.TableId, TableChange> tableSchemas) {
- if (!sourceConfig.isIncludeSchemaChanges()
- || tableSchemas == null
- || tableSchemas.isEmpty()) {
+ if (tableSchemas == null || tableSchemas.isEmpty()) {
return;
}
for (Map.Entry<io.debezium.relational.TableId, TableChange> entry :
@@ -160,7 +166,7 @@ public class SqlServerPipelineRecordEmitter<T> extends
IncrementalSourceRecordEm
if (tableId == null || tableChange == null ||
tableChange.getTable() == null) {
continue;
}
- createTableEventCache.putIfAbsent(
+ createTableEventCache.put(
tableId,
buildCreateTableEvent(
tableId,
SqlServerSchemaUtils.toSchema(tableChange.getTable())));
@@ -179,16 +185,19 @@ public class SqlServerPipelineRecordEmitter<T> extends
IncrementalSourceRecordEm
TableId.tableId(tableId.catalog(), tableId.schema(),
tableId.table()), schema);
}
- private void generateCreateTableEvents() {
+ private Map<io.debezium.relational.TableId, CreateTableEvent>
generateCreateTableEvents() {
try (SqlServerConnection jdbc =
createSqlServerConnection(sourceConfig.getDbzConnectorConfig())) {
+ Map<io.debezium.relational.TableId, CreateTableEvent>
createTableEvents =
+ new HashMap<>();
List<io.debezium.relational.TableId> capturedTableIds =
SqlServerConnectionUtils.listTables(
jdbc, sourceConfig.getTableFilters(),
sourceConfig.getDatabaseList());
for (io.debezium.relational.TableId tableId : capturedTableIds) {
CreateTableEvent createTableEvent =
buildCreateTableEvent(jdbc, tableId);
- createTableEventCache.put(tableId, createTableEvent);
+ createTableEvents.put(tableId, createTableEvent);
}
+ return createTableEvents;
} catch (SQLException e) {
throw new RuntimeException("Cannot start emitter to fetch table
schema.", e);
}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerPipelineSavepointRestoreITCase.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerPipelineSavepointRestoreITCase.java
new file mode 100644
index 000000000..51d863925
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerPipelineSavepointRestoreITCase.java
@@ -0,0 +1,428 @@
+/*
+ * 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.cdc.common.data.binary.BinaryRecordData;
+import org.apache.flink.cdc.common.data.binary.BinaryStringData;
+import org.apache.flink.cdc.common.event.AddColumnEvent;
+import org.apache.flink.cdc.common.event.AlterColumnTypeEvent;
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.DataChangeEvent;
+import org.apache.flink.cdc.common.event.DropColumnEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.common.schema.PhysicalColumn;
+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.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.configuration.Configuration;
+import org.apache.flink.configuration.StateRecoveryOptions;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.core.execution.SavepointFormatType;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.streaming.api.datastream.DataStreamSource;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.SinkFunction;
+import org.apache.flink.streaming.util.RestartStrategyUtils;
+import org.apache.flink.table.planner.factories.TestValuesTableFactory;
+import org.apache.flink.util.ExceptionUtils;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static
org.testcontainers.containers.MSSQLServerContainer.MS_SQL_SERVER_PORT;
+
+/** Savepoint/restore IT for SQL Server pipeline source schema evolution. */
+class SqlServerPipelineSavepointRestoreITCase extends SqlServerTestBase {
+
+ private static final String DATABASE_NAME = "customer";
+ private static final Map<String, BlockingQueue<Event>> EVENTS_BY_SINK =
+ new ConcurrentHashMap<>();
+
+ @BeforeEach
+ void before() {
+ initializeSqlServerTable(DATABASE_NAME);
+ TestValuesTableFactory.clearAllData();
+ EVENTS_BY_SINK.clear();
+ }
+
+ @Test
+ @Timeout(value = 300, unit = TimeUnit.SECONDS)
+ void testSchemaEvolutionContinuesAfterSavepointRestore() throws Exception {
+ TableId tableId = TableId.tableId(DATABASE_NAME, "dbo", "customers");
+ Schema schemaV1 = schemaV1();
+ Schema schemaV2 = schemaV2();
+ Schema schemaV3 = schemaV3();
+ Schema schemaV4 = schemaV4();
+
+ 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.customers")
+ .startupOptions(StartupOptions.initial())
+ .includeSchemaChanges(true)
+ .serverTimeZone("UTC");
+
+ String beforeSinkId = "before-" + UUID.randomUUID();
+ String afterSinkId = "after-" + UUID.randomUUID();
+ EVENTS_BY_SINK.put(beforeSinkId, new LinkedBlockingQueue<>());
+ EVENTS_BY_SINK.put(afterSinkId, new LinkedBlockingQueue<>());
+
+ Path savepointDir =
Files.createTempDirectory("sqlserver-schema-restore-test");
+ String savepointDirectory = savepointDir.toAbsolutePath().toString();
+
+ JobClient firstJobClient = null;
+ JobClient restoredJobClient = null;
+ try {
+ StreamExecutionEnvironment firstEnv =
getStreamExecutionEnvironment(null, 1);
+ DataStreamSource<Event> firstSource =
+ createSource(firstEnv, configFactory,
SqlServerDataSourceFactory.IDENTIFIER);
+ firstSource.addSink(new EventQueueSink(beforeSinkId)).name("Event
queue sink");
+ firstJobClient =
firstEnv.executeAsync("SqlServerSchemaRestoreBefore");
+
+ List<Event> firstRoundEvents = drainSinkResults(beforeSinkId, 22);
+ assertThat(
+ firstRoundEvents.stream()
+ .filter(CreateTableEvent.class::isInstance)
+ .collect(Collectors.toList()))
+ .containsExactly(new CreateTableEvent(tableId, schemaV1));
+ assertThat(
+ firstRoundEvents.stream()
+ .filter(event -> !(event instanceof
CreateTableEvent))
+ .collect(Collectors.toList()))
+
.containsExactlyInAnyOrderElementsOf(getSnapshotExpected(tableId, schemaV1));
+
+ Thread.sleep(2000L);
+ String savepointPath = triggerSavepointWithRetry(firstJobClient,
savepointDirectory);
+ firstJobClient.cancel().get();
+ firstJobClient = null;
+
+ StreamExecutionEnvironment restoredEnv =
+ getStreamExecutionEnvironment(savepointPath, 1);
+ DataStreamSource<Event> restoredSource =
+ createSource(restoredEnv, configFactory,
SqlServerDataSourceFactory.IDENTIFIER);
+ restoredSource.addSink(new
EventQueueSink(afterSinkId)).name("Event queue sink");
+ restoredJobClient =
restoredEnv.executeAsync("SqlServerSchemaRestoreAfter");
+
+ Thread.sleep(5000L);
+
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE " + DATABASE_NAME);
+ statement.execute("ALTER TABLE dbo.customers ADD ext INT");
+ statement.execute(
+ "EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', "
+ + "@source_name = 'customers', @role_name =
NULL, "
+ + "@supports_net_changes = 0,
@capture_instance = 'dbo_customers_v2';");
+ statement.execute(
+ "INSERT INTO dbo.customers VALUES "
+ + "(10000, 'Alice', 'Beijing', '123567891234',
17);");
+ }
+
+ assertThat(drainSinkResults(afterSinkId, 3))
+ .containsExactly(
+ new AddColumnEvent(
+ tableId,
+ Collections.singletonList(
+ new
AddColumnEvent.ColumnWithPosition(
+ new PhysicalColumn(
+ "ext",
DataTypes.INT(), null),
+
AddColumnEvent.ColumnPosition.AFTER,
+ "phone_number"))),
+ new CreateTableEvent(tableId, schemaV2),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(
+ schemaV2,
+ 10000,
+ "Alice",
+ "Beijing",
+ "123567891234",
+ 17)));
+
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE " + DATABASE_NAME);
+ statement.execute(
+ "EXEC sys.sp_cdc_disable_table @source_schema = 'dbo',
"
+ + "@source_name = 'customers', "
+ + "@capture_instance = 'dbo_customers';");
+ statement.execute("ALTER TABLE dbo.customers ALTER COLUMN ext
FLOAT");
+ statement.execute(
+ "EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', "
+ + "@source_name = 'customers', @role_name =
NULL, "
+ + "@supports_net_changes = 0,
@capture_instance = 'dbo_customers_v3';");
+ statement.execute(
+ "INSERT INTO dbo.customers VALUES "
+ + "(10001, 'Bob', 'Chongqing', '123567891234',
2.718281828);");
+ }
+
+ assertThat(drainSinkResults(afterSinkId, 2))
+ .containsExactly(
+ new AlterColumnTypeEvent(
+ tableId,
+ Collections.singletonMap("ext", (DataType)
DataTypes.DOUBLE()),
+ Collections.singletonMap("ext", (DataType)
DataTypes.INT()),
+ Collections.emptyMap()),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(
+ schemaV3,
+ 10001,
+ "Bob",
+ "Chongqing",
+ "123567891234",
+ 2.718281828)));
+
+ try (Connection connection = getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("USE " + DATABASE_NAME);
+ statement.execute(
+ "EXEC sys.sp_cdc_disable_table @source_schema = 'dbo',
"
+ + "@source_name = 'customers', "
+ + "@capture_instance = 'dbo_customers_v2';");
+ statement.execute("ALTER TABLE dbo.customers DROP COLUMN ext");
+ statement.execute(
+ "EXEC sys.sp_cdc_enable_table @source_schema = 'dbo', "
+ + "@source_name = 'customers', @role_name =
NULL, "
+ + "@supports_net_changes = 0,
@capture_instance = 'dbo_customers_v4';");
+ statement.execute(
+ "INSERT INTO dbo.customers VALUES "
+ + "(10002, 'Cicada', 'Urumqi',
'123567891234');");
+ }
+
+ assertThat(drainSinkResults(afterSinkId, 2))
+ .containsExactly(
+ new DropColumnEvent(tableId,
Collections.singletonList("ext")),
+ DataChangeEvent.insertEvent(
+ tableId,
+ generate(schemaV4, 10002, "Cicada",
"Urumqi", "123567891234")));
+ } finally {
+ if (restoredJobClient != null) {
+ restoredJobClient.cancel().get();
+ }
+ if (firstJobClient != null) {
+ firstJobClient.cancel().get();
+ }
+ }
+ }
+
+ private DataStreamSource<Event> createSource(
+ StreamExecutionEnvironment env,
+ SqlServerSourceConfigFactory configFactory,
+ String sourceName) {
+ FlinkSourceProvider sourceProvider =
+ (FlinkSourceProvider)
+ new
SqlServerDataSource(configFactory).getEventSourceProvider();
+ return env.fromSource(
+ sourceProvider.getSource(),
+ WatermarkStrategy.noWatermarks(),
+ sourceName,
+ new EventTypeInfo());
+ }
+
+ private List<Event> drainSinkResults(String sinkId, int size)
+ throws InterruptedException, TimeoutException {
+ BlockingQueue<Event> queue =
Objects.requireNonNull(EVENTS_BY_SINK.get(sinkId));
+ List<Event> result = new ArrayList<>(size);
+ for (int i = 0; i < size; i++) {
+ Event event = queue.poll(60L, TimeUnit.SECONDS);
+ if (event == null) {
+ throw new TimeoutException(
+ String.format(
+ "Timed out waiting for event %d/%d from sink
%s.",
+ i + 1, size, sinkId));
+ }
+ result.add(event);
+ }
+ return result;
+ }
+
+ private Schema schemaV1() {
+ return Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(),
null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+ }
+
+ private Schema schemaV2() {
+ return Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(),
null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .physicalColumn("ext", DataTypes.INT())
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+ }
+
+ private Schema schemaV3() {
+ return Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(),
null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .physicalColumn("ext", DataTypes.DOUBLE())
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+ }
+
+ private Schema schemaV4() {
+ return Schema.newBuilder()
+ .physicalColumn("id", DataTypes.INT().notNull())
+ .physicalColumn("name", DataTypes.VARCHAR(255).notNull(),
null, "flink")
+ .physicalColumn("address", DataTypes.VARCHAR(1024))
+ .physicalColumn("phone_number", DataTypes.VARCHAR(512))
+ .primaryKey(Collections.singletonList("id"))
+ .build();
+ }
+
+ private List<Event> getSnapshotExpected(TableId tableId, Schema schema) {
+ return Stream.of(
+ generate(schema, 101, "user_1", "Shanghai",
"123567891234"),
+ generate(schema, 102, "user_2", "Shanghai",
"123567891234"),
+ generate(schema, 103, "user_3", "Shanghai",
"123567891234"),
+ generate(schema, 109, "user_4", "Shanghai",
"123567891234"),
+ generate(schema, 110, "user_5", "Shanghai",
"123567891234"),
+ generate(schema, 111, "user_6", "Shanghai",
"123567891234"),
+ generate(schema, 118, "user_7", "Shanghai",
"123567891234"),
+ generate(schema, 121, "user_8", "Shanghai",
"123567891234"),
+ generate(schema, 123, "user_9", "Shanghai",
"123567891234"),
+ generate(schema, 1009, "user_10", "Shanghai",
"123567891234"),
+ generate(schema, 1010, "user_11", "Shanghai",
"123567891234"),
+ generate(schema, 1011, "user_12", "Shanghai",
"123567891234"),
+ generate(schema, 1012, "user_13", "Shanghai",
"123567891234"),
+ generate(schema, 1013, "user_14", "Shanghai",
"123567891234"),
+ generate(schema, 1014, "user_15", "Shanghai",
"123567891234"),
+ generate(schema, 1015, "user_16", "Shanghai",
"123567891234"),
+ generate(schema, 1016, "user_17", "Shanghai",
"123567891234"),
+ generate(schema, 1017, "user_18", "Shanghai",
"123567891234"),
+ generate(schema, 1018, "user_19", "Shanghai",
"123567891234"),
+ generate(schema, 1019, "user_20", "Shanghai",
"123567891234"),
+ generate(schema, 2000, "user_21", "Shanghai",
"123567891234"))
+ .map(record -> DataChangeEvent.insertEvent(tableId, record))
+ .collect(Collectors.toList());
+ }
+
+ private BinaryRecordData generate(Schema schema, Object... fields) {
+ BinaryRecordDataGenerator generator =
+ new
BinaryRecordDataGenerator(schema.getColumnDataTypes().toArray(new DataType[0]));
+ return generator.generate(
+ Stream.of(fields)
+ .map(
+ field ->
+ field instanceof String
+ ?
BinaryStringData.fromString((String) field)
+ : field)
+ .toArray());
+ }
+
+ private String triggerSavepointWithRetry(JobClient jobClient, String
savepointDirectory)
+ throws ExecutionException, InterruptedException {
+ int retryTimes = 0;
+ while (retryTimes < 600) {
+ try {
+ return jobClient
+ .triggerSavepoint(savepointDirectory,
SavepointFormatType.DEFAULT)
+ .get();
+ } catch (Exception e) {
+ if (ExceptionUtils.findThrowable(e, CheckpointException.class)
+ .filter(
+ exception ->
+ exception
+ .getMessage()
+ .contains("Checkpoint
triggering task"))
+ .isPresent()) {
+ Thread.sleep(100L);
+ retryTimes++;
+ } else {
+ throw e;
+ }
+ }
+ }
+ throw new AssertionError(
+ String.format(
+ "Failed to trigger savepoint in directory '%s' after
%d retries.",
+ savepointDirectory, retryTimes));
+ }
+
+ private StreamExecutionEnvironment getStreamExecutionEnvironment(
+ String savepointPath, int parallelism) {
+ Configuration configuration = new Configuration();
+ if (savepointPath != null) {
+ configuration.set(StateRecoveryOptions.SAVEPOINT_PATH,
savepointPath);
+ }
+ StreamExecutionEnvironment env =
+
StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+ env.setParallelism(parallelism);
+ env.enableCheckpointing(500L);
+ RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 3, 1000L);
+ return env;
+ }
+
+ private static class EventQueueSink implements SinkFunction<Event> {
+
+ private final String sinkId;
+
+ private EventQueueSink(String sinkId) {
+ this.sinkId = sinkId;
+ }
+
+ @Override
+ public void invoke(Event value, Context context) {
+ Objects.requireNonNull(EVENTS_BY_SINK.get(sinkId)).add(value);
+ }
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitterTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitterTest.java
new file mode 100644
index 000000000..ee5c0417a
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/reader/SqlServerPipelineRecordEmitterTest.java
@@ -0,0 +1,147 @@
+/*
+ * 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.reader;
+
+import org.apache.flink.cdc.common.event.CreateTableEvent;
+import org.apache.flink.cdc.common.event.Event;
+import org.apache.flink.cdc.common.schema.Schema;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
+import
org.apache.flink.cdc.connectors.sqlserver.source.SqlServerEventDeserializer;
+import
org.apache.flink.cdc.connectors.sqlserver.source.config.SqlServerSourceConfig;
+import org.apache.flink.cdc.connectors.sqlserver.utils.SqlServerSchemaUtils;
+import org.apache.flink.cdc.debezium.table.DebeziumChangelogMode;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+
+import io.debezium.config.Configuration;
+import io.debezium.relational.Column;
+import io.debezium.relational.Table;
+import io.debezium.relational.TableEditor;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Properties;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link SqlServerPipelineRecordEmitter}. */
+class SqlServerPipelineRecordEmitterTest {
+
+ private static final TableId DBZ_TABLE_ID = new TableId("db0", "dbo",
"users");
+ private static final org.apache.flink.cdc.common.event.TableId
CDC_TABLE_ID =
+ org.apache.flink.cdc.common.event.TableId.tableId("db0", "dbo",
"users");
+
+ @Test
+ void testApplySplitUsesRestoredSchemaOverExistingCache() {
+ SqlServerEventDeserializer deserializer =
+ new SqlServerEventDeserializer(DebeziumChangelogMode.ALL,
true);
+ deserializer
+ .getCreateTableEventCache()
+ .put(
+ DBZ_TABLE_ID,
+ new CreateTableEvent(
+ CDC_TABLE_ID,
+ schema(Collections.singletonList(col("id",
false, 1)))));
+
+ SqlServerPipelineRecordEmitter<Event> emitter =
+ new SqlServerPipelineRecordEmitter<>(
+ deserializer, null, createSourceConfig(), null, null);
+
+ emitter.applySplit(
+ new SnapshotSplit(
+ DBZ_TABLE_ID,
+ "db0.dbo.users:0",
+ RowType.of(new IntType()),
+ null,
+ null,
+ null,
+ Collections.singletonMap(
+ DBZ_TABLE_ID,
+ new TableChanges.TableChange(
+ TableChanges.TableChangeType.CREATE,
+ table(
+ Arrays.asList(
+ col("id", false, 1),
+ col("age", true,
2)))))));
+
+ assertThat(
+ deserializer
+ .getCreateTableEventCache()
+ .get(DBZ_TABLE_ID)
+ .getSchema()
+ .getColumnNames())
+ .containsExactly("id", "age");
+ }
+
+ private static SqlServerSourceConfig createSourceConfig() {
+ return new SqlServerSourceConfig(
+ StartupOptions.initial(),
+ Collections.singletonList("db0"),
+ Collections.singletonList("dbo.users"),
+ 8096,
+ 100,
+ 1000.0,
+ 0.05,
+ true,
+ false,
+ new Properties(),
+ Configuration.empty(),
+ "com.microsoft.sqlserver.jdbc.SQLServerDriver",
+ "localhost",
+ 1433,
+ "user",
+ "password",
+ 1024,
+ "UTC",
+ Duration.ofSeconds(30),
+ 3,
+ 1,
+ null,
+ false,
+ false,
+ false);
+ }
+
+ private static Schema schema(java.util.List<Column> columns) {
+ return SqlServerSchemaUtils.toSchema(table(columns));
+ }
+
+ private static Column col(String name, boolean optional, int position) {
+ return Column.editor()
+ .name(name)
+ .jdbcType(java.sql.Types.INTEGER)
+ .type("INT", "INT")
+ .position(position)
+ .optional(optional)
+ .create();
+ }
+
+ private static Table table(java.util.List<Column> columns) {
+ TableEditor editor = Table.editor().tableId(DBZ_TABLE_ID);
+ columns.forEach(editor::addColumn);
+ if (!columns.isEmpty()) {
+ editor.setPrimaryKeyNames("id");
+ }
+ return editor.create();
+ }
+}
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
index 402d51074..15666dc76 100644
---
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java
@@ -142,6 +142,9 @@ public class IncrementalSourceReader<T, C extends
SourceConfig>
@Override
protected SourceSplitState initializedState(SourceSplitBase split) {
+ if (recordEmitter instanceof IncrementalSourceRecordEmitter) {
+ ((IncrementalSourceRecordEmitter<?>)
recordEmitter).applySplit(split);
+ }
if (split.isSnapshotSplit()) {
return new SnapshotSplitState(split.asSnapshotSplit());
} else {
diff --git
a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReaderTest.java
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReaderTest.java
new file mode 100644
index 000000000..dc544e308
--- /dev/null
+++
b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/test/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReaderTest.java
@@ -0,0 +1,172 @@
+/*
+ * 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.base.source.reader;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.cdc.connectors.base.config.SourceConfig;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.Offset;
+import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceRecords;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
+import
org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitSerializer;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitState;
+import org.apache.flink.cdc.connectors.base.source.metrics.SourceReaderMetrics;
+import org.apache.flink.cdc.debezium.DebeziumDeserializationSchema;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
+import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
+import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Collector;
+
+import io.debezium.relational.TableId;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link IncrementalSourceReader}. */
+class IncrementalSourceReaderTest {
+
+ @Test
+ void testInitializedStateAppliesSplitToRecordEmitter() {
+ TestingReaderContext readerContext = new TestingReaderContext();
+ TrackingRecordEmitter recordEmitter =
+ new TrackingRecordEmitter(new
SourceReaderMetrics(readerContext.metricGroup()));
+ TestingIncrementalSourceReader reader =
+ new TestingIncrementalSourceReader(
+ recordEmitter, new
IncrementalSourceReaderContext(readerContext));
+
+ SnapshotSplit split =
+ new SnapshotSplit(
+ new TableId("catalog", "schema", "table"),
+ "split-0",
+ RowType.of(new IntType()),
+ null,
+ null,
+ null,
+ Collections.emptyMap());
+
+ reader.initialize(split);
+
+ assertThat(recordEmitter.appliedSplit).isSameAs(split);
+ }
+
+ private static class TestingIncrementalSourceReader
+ extends IncrementalSourceReader<Object, SourceConfig> {
+
+ private TestingIncrementalSourceReader(
+ TrackingRecordEmitter recordEmitter,
+ IncrementalSourceReaderContext incrementalSourceReaderContext)
{
+ super(
+ new
FutureCompletingBlockingQueue<RecordsWithSplitIds<SourceRecords>>(),
+ createSplitReaderSupplier(),
+ recordEmitter,
+ new Configuration(),
+ incrementalSourceReaderContext,
+ null,
+ new SourceSplitSerializer() {
+ @Override
+ public OffsetFactory getOffsetFactory() {
+ return TEST_OFFSET_FACTORY;
+ }
+ },
+ null);
+ }
+
+ private static Supplier<IncrementalSourceSplitReader<SourceConfig>>
+ createSplitReaderSupplier() {
+ return () -> null;
+ }
+
+ private SourceSplitState initialize(SourceSplitBase split) {
+ return initializedState(split);
+ }
+ }
+
+ private static class TrackingRecordEmitter extends
IncrementalSourceRecordEmitter<Object> {
+
+ private SourceSplitBase appliedSplit;
+
+ private TrackingRecordEmitter(SourceReaderMetrics sourceReaderMetrics)
{
+ super(
+ new NoOpDebeziumDeserializationSchema(),
+ sourceReaderMetrics,
+ false,
+ TEST_OFFSET_FACTORY);
+ }
+
+ @Override
+ public void applySplit(SourceSplitBase split) {
+ this.appliedSplit = split;
+ }
+ }
+
+ private static class NoOpDebeziumDeserializationSchema
+ implements DebeziumDeserializationSchema<Object> {
+
+ @Override
+ public void deserialize(SourceRecord record, Collector<Object> out) {
+ // no-op
+ }
+
+ @Override
+ public TypeInformation<Object> getProducedType() {
+ return TypeInformation.of(Object.class);
+ }
+ }
+
+ private static final OffsetFactory TEST_OFFSET_FACTORY =
+ new OffsetFactory() {
+ @Override
+ public Offset newOffset(Map<String, String> offset) {
+ return null;
+ }
+
+ @Override
+ public Offset newOffset(String filename, Long position) {
+ return null;
+ }
+
+ @Override
+ public Offset newOffset(Long position) {
+ return null;
+ }
+
+ @Override
+ public Offset createTimestampOffset(long timestampMillis) {
+ return null;
+ }
+
+ @Override
+ public Offset createInitialOffset() {
+ return null;
+ }
+
+ @Override
+ public Offset createNoStoppingOffset() {
+ return null;
+ }
+ };
+}
