[Fix][Connector-V2] Fix DB2 CDC capture and savepoint recovery

docs/en/connectors/source/DB2-CDC.md

@@ -195,6 +195,23 @@ source {
 }
 ```
 
+
+### Capture registrations and recovery
+
+Active Capture registrations remain readable as `CD_NEW_SYNCHPOINT` advances.
+`CD_OLD_SYNCHPOINT` is the readable starting point, not a stop watermark.
+Unquoted schema and table components in `table-names` and `table-names-config.table` are
+normalized to uppercase before catalog lookup and capture filtering when an object with the
+uppercase name exists. If only an object with the configured spelling exists, for example a
+table created with quoted lowercase identifiers that earlier releases matched without quotes,
+the configured spelling is kept. Quote a component to select one spelling explicitly. The
+database component retains its configured spelling, and `table-pattern` is kept unchanged.
+
+Savepoint recovery must preserve the schema component of DB2's two-part table identifiers.
+A restored job can catch up changes committed while it was stopped and discover a newly
+configured capture-enabled table. Existing configuration keys, offsets and checkpoint
+serialization formats remain unchanged.
+
 ## Changelog
 
 <ChangeLog />

docs/zh/connectors/source/DB2-CDC.md

@@ -193,6 +193,18 @@ source {
 }
 ```
 
+### Capture 注册与恢复
+
+活动 Capture 注册在 `CD_NEW_SYNCHPOINT` 推进后继续读取增量。
+`CD_OLD_SYNCHPOINT` 表示可读取的起点，不表示停止位点。
+`table-names` 和 `table-names-config.table` 中不带引号的 schema 和表名，在存在大写同名对象时会在
+catalog 查询与 Capture 过滤前转为大写；若只存在与配置拼写一致的对象（例如用带引号的小写标识符创建、
+早期版本中不带引号即可匹配的表），则保留配置中的拼写。需要明确选择某种拼写时请为该部分加引号。
+数据库名保留配置中的拼写，`table-pattern` 正则表达式保持原样。
+
+保存点恢复应保留 DB2 两段表标识中的 schema。恢复后的任务可以追平停机期间的变化，
+并发现新配置且已启用 Capture 的表。既有配置项、offset 和 checkpoint 序列化格式保持不变。
+
 ## 变更日志
 
 <ChangeLog />

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/reader/external/JdbcSourceFetchTaskContext.java

@@ -188,9 +188,13 @@ protected void registerDatabaseHistory(
                             jsonConverter.toConnectData("topic", historyTableChanges.get(tableId));
                     Struct deserializedStruct = (Struct) schemaAndValue.value();
 
+                    // The split key preserves the namespace even when history has only two parts.
+                    boolean useCatalogBeforeSchema =
+                            tableId.catalog() != null && !tableId.catalog().isEmpty();
                     TableChanges tableChanges =
                             tableChangeSerializer.deserialize(
-                                    Collections.singletonList(deserializedStruct), false);
+                                    Collections.singletonList(deserializedStruct),
+                                    useCatalogBeforeSchema);
 
                     Iterator<TableChanges.TableChange> iterator = tableChanges.iterator();
                     TableChanges.TableChange tableChange = null;

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/ConnectTableChangeSerializer.java

@@ -119,7 +119,11 @@ public TableChanges deserialize(List<Struct> data, boolean useCatalogBeforeSchem
             String tableId = struct.getString(ID_KEY);
             TableChanges.TableChangeType changeType =
                     TableChanges.TableChangeType.valueOf(struct.getString(TYPE_KEY));
-            Table table = toTable(struct.getStruct(TABLE_KEY), TableId.parse(tableId));
+            // Two-part identifiers mean catalog.table or schema.table depending on the caller.
+            Table table =
+                    toTable(
+                            struct.getStruct(TABLE_KEY),
+                            TableId.parse(tableId, useCatalogBeforeSchema));
             switch (changeType) {
                 case CREATE:
                     tableChanges.create(table);

seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/config/Db2SourceConfigFactory.java

@@ -77,30 +77,34 @@ public Db2SourceConfig create(int subtask) {
             dbzProperties.forEach(props::put);
         }
 
-        return new Db2SourceConfig(
-                startupConfig,
-                stopConfig,
-                databaseList,
-                tableList,
-                splitSize,
-                splitColumn,
-                distributionFactorUpper,
-                distributionFactorLower,
-                sampleShardingThreshold,
-                inverseSamplingRate,
-                sampleShardingAllow,
-                props,
-                DRIVER_CLASS_NAME,
-                hostname,
-                port,
-                username,
-                password,
-                originUrl,
-                fetchSize,
-                serverTimeZone,
-                connectTimeoutMillis,
-                connectMaxRetries,
-                connectionPoolSize,
-                exactlyOnce);
+        Db2SourceConfig config =
+                new Db2SourceConfig(
+                        startupConfig,
+                        stopConfig,
+                        databaseList,
+                        tableList,
+                        splitSize,
+                        splitColumn,
+                        distributionFactorUpper,
+                        distributionFactorLower,
+                        sampleShardingThreshold,
+                        inverseSamplingRate,
+                        sampleShardingAllow,
+                        props,
+                        DRIVER_CLASS_NAME,
+                        hostname,
+                        port,
+                        username,
+                        password,
+                        originUrl,
+                        fetchSize,
+                        serverTimeZone,
+                        connectTimeoutMillis,
+                        connectMaxRetries,
+                        connectionPoolSize,
+                        exactlyOnce);
+        // Preserve the user-selected snapshot splitting mode in each reader configuration.
+        config.setEnableConcurrentRead(enableConcurrentRead);
+        return config;
     }
 }

seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2Dialect.java

@@ -44,15 +44,18 @@
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges;
+import lombok.extern.slf4j.Slf4j;
 
 import java.sql.SQLException;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
 
 /** The {@link JdbcDataSourceDialect} implementation for Db2 datasource. */
+@Slf4j
 public class Db2Dialect implements JdbcDataSourceDialect {
 
     private static final long serialVersionUID = 1L;
@@ -160,13 +163,26 @@ public FetchTask<SourceSplitBase> createFetchTask(SourceSplitBase sourceSplitBas
 
     @Override
     public Optional<PrimaryKey> getPrimaryKey(JdbcConnection jdbcConnection, TableId tableId) {
-        return Optional.ofNullable(
-                tableMap.get(toDb2TableId(tableId)).getTableSchema().getPrimaryKey());
+        CatalogTable table = tableMap.get(toDb2TableId(tableId));
+        if (table == null) {
+            log.warn(
+                    "Catalog metadata not found for DB2 table {}; primary key is unavailable",
+                    tableId);
+            return Optional.empty();
+        }
+        return Optional.ofNullable(table.getTableSchema().getPrimaryKey());
     }
 
     @Override
     public List<ConstraintKey> getConstraintKeys(JdbcConnection jdbcConnection, TableId tableId) {
-        return tableMap.get(toDb2TableId(tableId)).getTableSchema().getConstraintKeys();
+        CatalogTable table = tableMap.get(toDb2TableId(tableId));
+        if (table == null) {
+            log.warn(
+                    "Catalog metadata not found for DB2 table {}; constraint keys are unavailable",
+                    tableId);
+            return Collections.emptyList();
+        }
+        return table.getTableSchema().getConstraintKeys();
     }
 
     private static Map<TableId, CatalogTable> createDb2TableMap(List<CatalogTable> catalogTables) {

seatunnel-connectors-v2/connector-cdc/connector-cdc-db2/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/db2/source/Db2IncrementalSourceFactory.java

@@ -17,29 +17,42 @@
 
 package org.apache.seatunnel.connectors.seatunnel.cdc.db2.source;
 
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
 import org.apache.seatunnel.api.configuration.util.OptionRule;
 import org.apache.seatunnel.api.options.ConnectorCommonOptions;
 import org.apache.seatunnel.api.source.SeaTunnelSource;
 import org.apache.seatunnel.api.source.SourceSplit;
+import org.apache.seatunnel.api.table.catalog.Catalog;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.CatalogTableUtil;
 import org.apache.seatunnel.api.table.catalog.TablePath;
 import org.apache.seatunnel.api.table.connector.TableSource;
 import org.apache.seatunnel.api.table.factory.Factory;
+import org.apache.seatunnel.api.table.factory.FactoryUtil;
 import org.apache.seatunnel.api.table.factory.TableSourceFactory;
 import org.apache.seatunnel.api.table.factory.TableSourceFactoryContext;
 import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceTableConfig;
 import org.apache.seatunnel.connectors.cdc.base.option.SourceOptions;
 import org.apache.seatunnel.connectors.cdc.base.option.StartupMode;
 import org.apache.seatunnel.connectors.cdc.base.option.StopMode;
 import org.apache.seatunnel.connectors.cdc.base.utils.CatalogTableUtils;
+import org.apache.seatunnel.connectors.seatunnel.cdc.db2.utils.TableDiscoveryUtils;
+import org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.DatabaseIdentifier;
 
 import com.google.auto.service.AutoService;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.function.Function;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.apache.seatunnel.api.configuration.util.Conditions.notBlank;
 
@@ -114,20 +127,113 @@ TableSource<T, SplitT, StateT> createSource(TableSourceFactoryContext context) {
             } catch (Exception e) {
                 log.warn("Failed to load JDBC driver {}", DRIVER_CLASS_NAME, e);
             }
+            ReadonlyConfig options =
+                    resolveDb2Options(context.getOptions(), context.getClassLoader());
             List<CatalogTable> catalogTables =
-                    CatalogTableUtil.getCatalogTables(
-                            context.getOptions(), context.getClassLoader());
+                    CatalogTableUtil.getCatalogTables(options, context.getClassLoader());
             Optional<List<JdbcSourceTableConfig>> tableConfigs =
-                    context.getOptions()
-                            .getOptional(Db2IncrementalSourceOptions.TABLE_NAMES_CONFIG);
+                    options.getOptional(Db2IncrementalSourceOptions.TABLE_NAMES_CONFIG);
             if (tableConfigs.isPresent()) {
                 catalogTables =
                         CatalogTableUtils.mergeCatalogTableConfig(
                                 catalogTables,
                                 tableConfigs.get(),
                                 text -> TablePath.of(text, true));
             }
-            return new Db2IncrementalSource(context.getOptions(), catalogTables);
+            return new Db2IncrementalSource(options, catalogTables);
         };
     }
+
+    /**
+     * Resolves configured table names against the DB2 catalog before they are used anywhere.
+     *
+     * <p>The catalog is opened only when folding would change at least one unquoted name. That
+     * keeps uppercase and quoted configurations on the previous single-connection path, while
+     * lowercase unquoted names get one additional short-lived connection to decide between the
+     * folded spelling and the configured spelling. Without a DB2 catalog factory the names are
+     * folded unconditionally, which is the behavior before this fallback existed.
+     */
+    static ReadonlyConfig resolveDb2Options(ReadonlyConfig options, ClassLoader classLoader) {
+        boolean foldingChangesAnyName =
+                Stream.concat(
+                                options.getOptional(ConnectorCommonOptions.TABLE_NAMES)
+                                        .map(List::stream)
+                                        .orElseGet(Stream::empty),
+                                options.getOptional(Db2IncrementalSourceOptions.TABLE_NAMES_CONFIG)
+                                        .map(
+                                                tables ->
+                                                        tables.stream()
+                                                                .map(
+                                                                        JdbcSourceTableConfig
+                                                                                ::getTable))
+                                        .orElseGet(Stream::empty))
+                        .anyMatch(
+                                name ->
+                                        !TableDiscoveryUtils.normalizeConfiguredTableName(name)
+                                                .equals(name));
+        if (!foldingChangesAnyName) {
+            return normalizeDb2Options(options, tablePath -> true);
+        }
+        Optional<Catalog> optionalCatalog =
+                FactoryUtil.createOptionalCatalog(
+                        DatabaseIdentifier.DB_2, options, classLoader, DatabaseIdentifier.DB_2);
+        if (!optionalCatalog.isPresent()) {
+            return normalizeDb2Options(options, tablePath -> true);
+        }
+        try (Catalog catalog = optionalCatalog.get()) {
+            catalog.open();
+            return normalizeDb2Options(options, catalog::tableExists);
+        }
+    }
+
+    /**
+     * Keeps catalog lookup, per-table overrides and capture filters in the same identifier space.
+     * Each distinct configured name is resolved once through {@code tableExists}; see {@link
+     * TableDiscoveryUtils#resolveConfiguredTableName(String, Predicate)} for the precedence. Table
+     * patterns are regular expressions and must not be case-folded here.
+     */
+    static ReadonlyConfig normalizeDb2Options(
+            ReadonlyConfig options, Predicate<TablePath> tableExists) {
+        Map<String, String> resolvedNames = new HashMap<>();
+        Function<String, String> resolver =
+                name ->
+                        resolvedNames.computeIfAbsent(
+                                name,
+                                configuredTable ->
+                                        TableDiscoveryUtils.resolveConfiguredTableName(
+                                                configuredTable, tableExists));
+        Map<String, Object> normalized = new LinkedHashMap<>(options.getSourceMap());
+        options.getOptional(ConnectorCommonOptions.TABLE_NAMES)
+                .ifPresent(
+                        names ->
+                                normalized.put(
+                                        ConnectorCommonOptions.TABLE_NAMES.key(),
+                                        names.stream().map(resolver).collect(Collectors.toList())));
+        options.getOptional(Db2IncrementalSourceOptions.TABLE_NAMES_CONFIG)
+                .ifPresent(
+                        tables ->
+                                normalized.put(
+                                        Db2IncrementalSourceOptions.TABLE_NAMES_CONFIG.key(),
+                                        tables.stream()
+                                                .map(table -> normalizeTableConfig(table, resolver))
+                                                .collect(Collectors.toList())));
+        return ReadonlyConfig.fromMap(normalized);
+    }
+
+    /**
+     * Preserves user-configured primary keys and split-column override while resolving the table
+     * name with the same resolver as {@code table-names}.
+     */
+    private static Map<String, Object> normalizeTableConfig(
+            JdbcSourceTableConfig config, Function<String, String> resolver) {
+        Map<String, Object> normalized = new LinkedHashMap<>();
+        normalized.put("table", resolver.apply(config.getTable()));
+        if (config.getPrimaryKeys() != null) {
+            normalized.put("primaryKeys", new ArrayList<>(config.getPrimaryKeys()));
+        }
+        if (config.getSnapshotSplitColumn() != null) {
+            normalized.put("snapshotSplitColumn", config.getSnapshotSplitColumn());
+        }
+        return normalized;
+    }
 }