[feature][connector-v2] add sqlServer CDC (#3686)

* [feature][connector-v2] add sqlServer CDC

* [feature][connector-v2] add sqlServer CDC E2E

Author: ic4y

plugin-mapping.properties

@@ -160,6 +160,7 @@ seatunnel.source.Notion = connector-http-notion
 seatunnel.sink.RabbitMQ = connector-rabbitmq
 seatunnel.source.RabbitMQ = connector-rabbitmq
 seatunnel.source.OpenMldb = connector-openmldb
+seatunnel.source.SqlServer-CDC = connector-cdc-sqlserver
 seatunnel.sink.Doris = connector-doris
 seatunnel.source.Maxcompute = connector-maxcompute
 seatunnel.sink.Maxcompute = connector-maxcompute

seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/base/source/IncrementalSource.java

@@ -85,8 +85,8 @@ public final void prepare(Config pluginConfig) throws PrepareFailException {
         this.stopMode = stopConfig.getStopMode();
         this.incrementalParallelism = readonlyConfig.get(SourceOptions.INCREMENTAL_PARALLELISM);
         this.configFactory = createSourceConfigFactory(readonlyConfig);
-        this.deserializationSchema = createDebeziumDeserializationSchema(readonlyConfig);
         this.dataSourceDialect = createDataSourceDialect(readonlyConfig);
+        this.deserializationSchema = createDebeziumDeserializationSchema(readonlyConfig);
         this.offsetFactory = createOffsetFactory(readonlyConfig);
     }
 

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

@@ -334,7 +334,7 @@ public Object convert(Object dbzObj, Schema schema) {
                         default:
                     }
                 } else if (dbzObj instanceof Integer) {
-                    return LocalTime.ofNanoOfDay((long) dbzObj * 1000_000L);
+                    return LocalTime.ofNanoOfDay((Integer) dbzObj * 1000_000L);
                 }
                 // get number of milliseconds of the day
                 return TemporalConversions.toLocalTime(dbzObj);

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/MySqlSourceFetchTaskContext.java

@@ -235,13 +235,12 @@ public Offset getStreamOffset(SourceRecord sourceRecord) {
      * Loads the connector's persistent offset (if present) via the given loader.
      */
     private MySqlOffsetContext loadStartingOffsetState(
-        OffsetContext.Loader loader, SourceSplitBase mySqlSplit) {
+        MySqlOffsetContext.Loader loader, SourceSplitBase mySqlSplit) {
         Offset offset =
             mySqlSplit.isSnapshotSplit() ? BinlogOffset.INITIAL_OFFSET
                 : mySqlSplit.asIncrementalSplit().getStartupOffset();
 
-        MySqlOffsetContext mySqlOffsetContext =
-            (MySqlOffsetContext) loader.load(offset.getOffset());
+        MySqlOffsetContext mySqlOffsetContext = loader.load(offset.getOffset());
 
         if (!isBinlogAvailable(mySqlOffsetContext)) {
             throw new IllegalStateException(

seatunnel-connectors-v2/connector-cdc/connector-cdc-mysql/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/source/reader/fetch/scan/MySqlSnapshotSplitReadTask.java

@@ -31,15 +31,13 @@
 import io.debezium.connector.mysql.MySqlConnectorConfig;
 import io.debezium.connector.mysql.MySqlDatabaseSchema;
 import io.debezium.connector.mysql.MySqlOffsetContext;
-import io.debezium.connector.mysql.MySqlValueConverters;
 import io.debezium.pipeline.EventDispatcher;
 import io.debezium.pipeline.source.AbstractSnapshotChangeEventSource;
 import io.debezium.pipeline.source.spi.ChangeEventSource;
 import io.debezium.pipeline.source.spi.SnapshotProgressListener;
 import io.debezium.pipeline.spi.ChangeRecordEmitter;
 import io.debezium.pipeline.spi.OffsetContext;
 import io.debezium.pipeline.spi.SnapshotResult;
-import io.debezium.relational.Column;
 import io.debezium.relational.RelationalSnapshotChangeEventSource;
 import io.debezium.relational.SnapshotChangeRecordEmitter;
 import io.debezium.relational.Table;
@@ -52,14 +50,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.io.UnsupportedEncodingException;
-import java.sql.Blob;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Types;
 import java.time.Duration;
-import java.util.Calendar;
 
 public class MySqlSnapshotSplitReadTask extends AbstractSnapshotChangeEventSource {
 
@@ -146,7 +142,7 @@ protected SnapshotResult doExecute(
             "Snapshot step 3 - Determining high watermark {} for split {}",
             highWatermark,
             snapshotSplit);
-        ((SnapshotSplitChangeEventSourceContext) context).setHighWatermark(lowWatermark);
+        ((SnapshotSplitChangeEventSourceContext) context).setHighWatermark(highWatermark);
         dispatcher.dispatchWatermarkEvent(
             offsetContext.getPartition(), snapshotSplit, highWatermark, WatermarkKind.HIGH);
         return SnapshotResult.completed(ctx.offset);
@@ -220,9 +216,8 @@ private void createDataEventsForTable(
                 rows++;
                 final Object[] row = new Object[columnArray.getGreatestColumnPosition()];
                 for (int i = 0; i < columnArray.getColumns().length; i++) {
-                    Column actualColumn = table.columns().get(i);
                     row[columnArray.getColumns()[i].position() - 1] =
-                        readField(rs, i + 1, actualColumn, table);
+                        readField(rs, i + 1);
                 }
                 if (logTimer.expired()) {
                     long stop = clock.currentTimeInMillis();
@@ -259,108 +254,15 @@ private Threads.Timer getTableScanLogTimer() {
         return Threads.timer(clock, LOG_INTERVAL);
     }
 
-    /**
-     * Read JDBC return value and deal special type like time, timestamp.
-     *
-     * <p>Note https://issues.redhat.com/browse/DBZ-3238 has fixed this issue, please remove
-     * this method once we bump Debezium version to 1.6
-     */
-    private Object readField(ResultSet rs, int fieldNo, Column actualColumn, Table actualTable)
+    private Object readField(
+        ResultSet rs, int columnIndex)
         throws SQLException {
-        if (actualColumn.jdbcType() == Types.TIME) {
-            return readTimeField(rs, fieldNo);
-        } else if (actualColumn.jdbcType() == Types.DATE) {
-            return readDateField(rs, fieldNo, actualColumn, actualTable);
-        }
-        // This is for DATETIME columns (a logical date + time without time zone)
-        // by reading them with a calendar based on the default time zone, we make sure that the
-        // value
-        // is constructed correctly using the database's (or connection's) time zone
-        else if (actualColumn.jdbcType() == Types.TIMESTAMP) {
-            return readTimestampField(rs, fieldNo, actualColumn, actualTable);
-        }
-        // JDBC's rs.GetObject() will return a Boolean for all TINYINT(1) columns.
-        // TINYINT columns are reprtoed as SMALLINT by JDBC driver
-        else if (actualColumn.jdbcType() == Types.TINYINT
-            || actualColumn.jdbcType() == Types.SMALLINT) {
-            // It seems that rs.wasNull() returns false when default value is set and NULL is
-            // inserted
-            // We thus need to use getObject() to identify if the value was provided and if yes
-            // then
-            // read it again to get correct scale
-            return rs.getObject(fieldNo) == null ? null : rs.getInt(fieldNo);
-        }
-        // DBZ-2673
-        // It is necessary to check the type names as types like ENUM and SET are
-        // also reported as JDBC type char
-        else if ("CHAR".equals(actualColumn.typeName())
-            || "VARCHAR".equals(actualColumn.typeName())
-            || "TEXT".equals(actualColumn.typeName())) {
-            return rs.getBytes(fieldNo);
+        final ResultSetMetaData metaData = rs.getMetaData();
+        final int columnType = metaData.getColumnType(columnIndex);
+        if (columnType == Types.TIME) {
+            return rs.getTimestamp(columnIndex);
         } else {
-            return rs.getObject(fieldNo);
-        }
-    }
-
-    /**
-     * As MySQL connector/J implementation is broken for MySQL type "TIME" we have to use a
-     * binary-ish workaround. https://issues.jboss.org/browse/DBZ-342
-     */
-    private Object readTimeField(ResultSet rs, int fieldNo) throws SQLException {
-        Blob b = rs.getBlob(fieldNo);
-        if (b == null) {
-            return null; // Don't continue parsing time field if it is null
-        }
-
-        try {
-            return MySqlValueConverters.stringToDuration(
-                new String(b.getBytes(1, (int) (b.length())), "UTF-8"));
-        } catch (UnsupportedEncodingException e) {
-            LOG.error("Could not read MySQL TIME value as UTF-8");
-            throw new RuntimeException(e);
-        }
-    }
-
-    /**
-     * In non-string mode the date field can contain zero in any of the date part which we need
-     * to handle as all-zero.
-     */
-    private Object readDateField(ResultSet rs, int fieldNo, Column column, Table table)
-        throws SQLException {
-        Blob b = rs.getBlob(fieldNo);
-        if (b == null) {
-            return null; // Don't continue parsing date field if it is null
-        }
-
-        try {
-            return MySqlValueConverters.stringToLocalDate(
-                new String(b.getBytes(1, (int) (b.length())), "UTF-8"), column, table);
-        } catch (UnsupportedEncodingException e) {
-            LOG.error("Could not read MySQL TIME value as UTF-8");
-            throw new RuntimeException(e);
-        }
-    }
-
-    /**
-     * In non-string mode the time field can contain zero in any of the date part which we need
-     * to handle as all-zero.
-     */
-    private Object readTimestampField(ResultSet rs, int fieldNo, Column column, Table table)
-        throws SQLException {
-        Blob b = rs.getBlob(fieldNo);
-        if (b == null) {
-            return null; // Don't continue parsing timestamp field if it is null
-        }
-
-        try {
-            return MySqlValueConverters.containsZeroValuesInDatePart(
-                new String(b.getBytes(1, (int) (b.length())), "UTF-8"),
-                column,
-                table) ? null
-                : rs.getTimestamp(fieldNo, Calendar.getInstance());
-        } catch (UnsupportedEncodingException e) {
-            LOG.error("Could not read MySQL TIME value as UTF-8");
-            throw new RuntimeException(e);
+            return rs.getObject(columnIndex);
         }
     }
 

seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/pom.xml

@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>connector-cdc</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>connector-cdc-sqlserver</artifactId>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-cdc-base</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.debezium</groupId>
+            <artifactId>debezium-connector-sqlserver</artifactId>
+        </dependency>
+    </dependencies>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>org.apache.seatunnel</groupId>
+                <artifactId>connector-cdc-base</artifactId>
+                <version>${project.version}</version>
+                <scope>compile</scope>
+            </dependency>
+
+            <dependency>
+                <groupId>io.debezium</groupId>
+                <artifactId>debezium-connector-sqlserver</artifactId>
+                <version>${debezium.version}</version>
+                <scope>compile</scope>
+            </dependency>
+
+        </dependencies>
+    </dependencyManagement>
+
+</project>

seatunnel-connectors-v2/connector-cdc/connector-cdc-sqlserver/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/sqlserver/source/config/SqlServerSourceConfig.java

@@ -0,0 +1,87 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.cdc.sqlserver.source.config;
+
+import org.apache.seatunnel.connectors.cdc.base.config.JdbcSourceConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.StartupConfig;
+import org.apache.seatunnel.connectors.cdc.base.config.StopConfig;
+
+import io.debezium.connector.sqlserver.SqlServerConnectorConfig;
+import io.debezium.relational.RelationalTableFilters;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Describes the connection information of the Mysql database and the configuration information for
+ * performing snapshotting and streaming reading, such as splitSize.
+ */
+public class SqlServerSourceConfig extends JdbcSourceConfig {
+
+    private static final long serialVersionUID = 1L;
+
+    public SqlServerSourceConfig(
+            StartupConfig startupConfig,
+            StopConfig stopConfig,
+            List<String> databaseList,
+            List<String> tableList,
+            int splitSize,
+            double distributionFactorUpper,
+            double distributionFactorLower,
+            Properties dbzProperties,
+            String driverClassName,
+            String hostname,
+            int port,
+            String username,
+            String password,
+            int fetchSize,
+            String serverTimeZone,
+            Duration connectTimeout,
+            int connectMaxRetries,
+            int connectionPoolSize) {
+        super(
+                startupConfig,
+                stopConfig,
+                databaseList,
+                tableList,
+                splitSize,
+                distributionFactorUpper,
+                distributionFactorLower,
+                dbzProperties,
+                driverClassName,
+                hostname,
+                port,
+                username,
+                password,
+                fetchSize,
+                serverTimeZone,
+                connectTimeout,
+                connectMaxRetries,
+                connectionPoolSize);
+    }
+
+    @Override
+    public SqlServerConnectorConfig getDbzConnectorConfig() {
+        return new SqlServerConnectorConfig(getDbzConfiguration());
+    }
+
+    public RelationalTableFilters getTableFilters() {
+        return getDbzConnectorConfig().getTableFilters();
+    }
+}