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 c45eb567d [hotfix][pipeline-connector/sqlserver] Align pipeline
default value with SQL API
c45eb567d is described below
commit c45eb567d87582d50b886eb30803390fae1304d0
Author: Leonard Xu <[email protected]>
AuthorDate: Thu Jul 23 16:21:24 2026 +0800
[hotfix][pipeline-connector/sqlserver] Align pipeline default value with
SQL API
This closes #4485.
---
.../connectors/pipeline-connectors/sqlserver.md | 11 +++++++++-
.../connectors/pipeline-connectors/sqlserver.md | 11 +++++++++-
.../source/SqlServerDataSourceOptions.java | 6 +++---
.../factory/SqlServerDataSourceFactoryTest.java | 25 ++++++++++++++++++++++
4 files changed, 48 insertions(+), 5 deletions(-)
diff --git a/docs/content.zh/docs/connectors/pipeline-connectors/sqlserver.md
b/docs/content.zh/docs/connectors/pipeline-connectors/sqlserver.md
index 74ae6e1db..b45bd8676 100644
--- a/docs/content.zh/docs/connectors/pipeline-connectors/sqlserver.md
+++ b/docs/content.zh/docs/connectors/pipeline-connectors/sqlserver.md
@@ -198,10 +198,17 @@ pipeline:
<tr>
<td>scan.incremental.snapshot.backfill.skip</td>
<td>optional</td>
- <td style="word-wrap: break-word;">true</td>
+ <td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>是否在快照读取阶段跳过 backfill。跳过 backfill 可能导致部分 change log 事件以 at-least-once
语义被重放。</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>是否在快照读取阶段优先分配无上界 chunk。这可能有助于降低对最大的无上界 chunk 执行快照时 TaskManager
出现内存溢出（OOM）的风险。</td>
+ </tr>
<tr>
<td>connect.timeout</td>
<td>optional</td>
@@ -262,6 +269,8 @@ pipeline:
</table>
</div>
+> 兼容性说明：如果希望保持旧版本 Pipeline 行为，可以显式配置 `scan.incremental.snapshot.backfill.skip:
true` 和 `scan.incremental.snapshot.unbounded-chunk-first.enabled: false`。
+
## 可用 Metadata
配置 `metadata.list` 后，以下 metadata 可以传递到下游。
diff --git a/docs/content/docs/connectors/pipeline-connectors/sqlserver.md
b/docs/content/docs/connectors/pipeline-connectors/sqlserver.md
index 3ee6b5ff3..10345d3bf 100644
--- a/docs/content/docs/connectors/pipeline-connectors/sqlserver.md
+++ b/docs/content/docs/connectors/pipeline-connectors/sqlserver.md
@@ -199,10 +199,17 @@ pipeline:
<tr>
<td>scan.incremental.snapshot.backfill.skip</td>
<td>optional</td>
- <td style="word-wrap: break-word;">true</td>
+ <td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>Whether to skip backfill in snapshot reading phase. Skipping
backfill may lead to replayed change log events with at-least-once
semantics.</td>
</tr>
+ <tr>
+ <td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
+ <td>optional</td>
+ <td style="word-wrap: break-word;">true</td>
+ <td>Boolean</td>
+ <td>Whether to assign the unbounded chunks first during snapshot reading
phase. This might help reduce the risk of the TaskManager experiencing an
out-of-memory (OOM) error when taking a snapshot of the largest unbounded
chunk.</td>
+ </tr>
<tr>
<td>connect.timeout</td>
<td>optional</td>
@@ -263,6 +270,8 @@ pipeline:
</table>
</div>
+> Compatibility note: To keep the behavior of older Pipeline versions,
explicitly configure `scan.incremental.snapshot.backfill.skip: true` and
`scan.incremental.snapshot.unbounded-chunk-first.enabled: false`.
+
## Available Metadata
The following metadata can be passed downstream when configured in
`metadata.list`.
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/SqlServerDataSourceOptions.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerDataSourceOptions.java
index bf21ca22e..1b409ebbf 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerDataSourceOptions.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/main/java/org/apache/flink/cdc/connectors/sqlserver/source/SqlServerDataSourceOptions.java
@@ -136,7 +136,7 @@ public class SqlServerDataSourceOptions {
public static final ConfigOption<Boolean>
SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP =
ConfigOptions.key("scan.incremental.snapshot.backfill.skip")
.booleanType()
- .defaultValue(true)
+ .defaultValue(false)
.withDescription(
"Whether to skip backfill in snapshot reading
phase. If backfill is skipped, changes on captured tables during snapshot phase
will be consumed later in change log reading phase instead of being merged into
the snapshot.WARNING: Skipping backfill might lead to data inconsistency
because some change log events happened within the snapshot phase might be
replayed (only at-least-once semantic is promised). For example updating an
already updated value in snapshot, [...]
@@ -223,9 +223,9 @@ public class SqlServerDataSourceOptions {
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED =
ConfigOptions.key("scan.incremental.snapshot.unbounded-chunk-first.enabled")
.booleanType()
- .defaultValue(false)
+ .defaultValue(true)
.withDescription(
- "Whether to assign the unbounded chunks
first during snapshot reading phase. This might help reduce the risk of the
TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of
the largest unbounded chunk. Defaults to false.");
+ "Whether to assign the unbounded chunks
first during snapshot reading phase. This might help reduce the risk of the
TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of
the largest unbounded chunk.");
@Experimental
public static final ConfigOption<Boolean> SCAN_NEWLY_ADDED_TABLE_ENABLED =
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactoryTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactoryTest.java
index a723e7288..51d891537 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactoryTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-sqlserver/src/test/java/org/apache/flink/cdc/connectors/sqlserver/factory/SqlServerDataSourceFactoryTest.java
@@ -39,7 +39,9 @@ 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_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_UNBOUNDED_CHUNK_FIRST_ENABLED;
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.TABLES;
@@ -76,6 +78,29 @@ public class SqlServerDataSourceFactoryTest extends
SqlServerTestBase {
.isEqualTo(Arrays.asList("dbo.products",
"dbo.products_on_hand"));
}
+ @Test
+ public void testSnapshotBackfillAndUnboundedChunkFirstDefaults() {
+ Map<String, String> options = new HashMap<>();
+ options.put(HOSTNAME.key(), MSSQL_SERVER_CONTAINER.getHost());
+ options.put(
+ PORT.key(),
+
String.valueOf(MSSQL_SERVER_CONTAINER.getMappedPort(MS_SQL_SERVER_PORT)));
+ options.put(USERNAME.key(), MSSQL_SERVER_CONTAINER.getUsername());
+ options.put(PASSWORD.key(), MSSQL_SERVER_CONTAINER.getPassword());
+ options.put(TABLES.key(), DATABASE_NAME + ".dbo.prod\\.*");
+
+ Factory.Context context = new
MockContext(Configuration.fromMap(options));
+ SqlServerDataSourceFactory factory = new SqlServerDataSourceFactory();
+
+ assertThat(factory.optionalOptions())
+ .contains(
+ SCAN_INCREMENTAL_SNAPSHOT_BACKFILL_SKIP,
+
SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST_ENABLED);
+ SqlServerDataSource dataSource = (SqlServerDataSource)
factory.createDataSource(context);
+
assertThat(dataSource.getSqlServerSourceConfig().isSkipSnapshotBackfill()).isFalse();
+
assertThat(dataSource.getSqlServerSourceConfig().isAssignUnboundedChunkFirst()).isTrue();
+ }
+
@Test
public void testNoMatchedTable() {
Map<String, String> options = new HashMap<>();
