[Feature][Connector-V2][SqlServer] Support driver jtds for SqlServer #5307

docs/en/connector-v2/sink/Jdbc.md

@@ -170,6 +170,7 @@ there are some reference value for params above.
 | Snowflake  | net.snowflake.client.jdbc.SnowflakeDriver    | jdbc:snowflake://<account_name>.snowflakecomputing.com             | /                                                  | https://mvnrepository.com/artifact/net.snowflake/snowflake-jdbc                                             |
 | Vertica    | com.vertica.jdbc.Driver                      | jdbc:vertica://localhost:5433                                      | /                                                  | https://repo1.maven.org/maven2/com/vertica/jdbc/vertica-jdbc/12.0.3-0/vertica-jdbc-12.0.3-0.jar             |
 | OceanBase  | com.oceanbase.jdbc.Driver                    | jdbc:oceanbase://localhost:2881                                    | /                                                  | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.3/oceanbase-client-2.4.3.jar              |
+| sqlserver  | net.sourceforge.jtds.jdbc.Driver             | jdbc:jtds:sqlserver://localhost:1433                               | /                                                  | https://mvnrepository.com/artifact/net.sourceforge.jtds/jtds/1.3.1/jtds-1.3.1.jar                           |
 
 ## Example
 

docs/en/connector-v2/source/Jdbc.md

@@ -126,6 +126,7 @@ there are some reference value for params above.
 | Redshift   | com.amazon.redshift.jdbc42.Driver                   | jdbc:redshift://localhost:5439/testdb?defaultRowFetchSize=1000         | https://mvnrepository.com/artifact/com.amazon.redshift/redshift-jdbc42                                      |
 | Vertica    | com.vertica.jdbc.Driver                             | jdbc:vertica://localhost:5433                                          | https://repo1.maven.org/maven2/com/vertica/jdbc/vertica-jdbc/12.0.3-0/vertica-jdbc-12.0.3-0.jar             |
 | OceanBase  | com.oceanbase.jdbc.Driver                           | jdbc:oceanbase://localhost:2881                                        | https://repo1.maven.org/maven2/com/oceanbase/oceanbase-client/2.4.3/oceanbase-client-2.4.3.jar              |
+| sqlserver  | net.sourceforge.jtds.jdbc.Driver                    | jdbc:jtds:sqlserver://localhost:1433                                   | https://mvnrepository.com/artifact/net.sourceforge.jtds/jtds/1.3.1/jtds-1.3.1.jar                           |
 
 ## Example
 

seatunnel-connectors-v2/connector-jdbc/pom.xml

@@ -34,6 +34,7 @@
         <postgresql.version>42.4.3</postgresql.version>
         <dm-jdbc.version>8.1.2.141</dm-jdbc.version>
         <sqlserver.version>9.2.1.jre8</sqlserver.version>
+        <jtds.version>1.3.1</jtds.version>
         <phoenix.version>5.2.5-HBase-2.x</phoenix.version>
         <oracle.version>12.2.0.1</oracle.version>
         <sqlite.version>3.39.3.0</sqlite.version>
@@ -86,6 +87,12 @@
                 <version>${sqlserver.version}</version>
                 <scope>provided</scope>
             </dependency>
+            <dependency>
+                <groupId>net.sourceforge.jtds</groupId>
+                <artifactId>jtds</artifactId>
+                <version>${jtds.version}</version>
+                <scope>provided</scope>
+            </dependency>
             <dependency>
                 <groupId>com.oracle.database.jdbc</groupId>
                 <artifactId>ojdbc8</artifactId>
@@ -180,6 +187,12 @@
             <groupId>com.microsoft.sqlserver</groupId>
             <artifactId>mssql-jdbc</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>net.sourceforge.jtds</groupId>
+            <artifactId>jtds</artifactId>
+        </dependency>
+
         <dependency>
             <groupId>com.oracle.database.jdbc</groupId>
             <artifactId>ojdbc8</artifactId>

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/connection/SimpleJdbcConnectionProvider.java

@@ -59,8 +59,12 @@ public Connection getConnection() {
 
     @Override
     public boolean isConnectionValid() throws SQLException {
-        return connection != null
-                && connection.isValid(jdbcConfig.getConnectionCheckTimeoutSeconds());
+        if (connection != null && connection.toString().startsWith("net.sourceforge.jtds")) {
+            return connection != null && !connection.isClosed();
+        } else {
+            return connection != null
+                    && connection.isValid(jdbcConfig.getConnectionCheckTimeoutSeconds());
+        }
     }
 
     private static Driver loadDriver(String driverName) throws ClassNotFoundException {

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlServerDialectFactory.java

@@ -27,7 +27,7 @@
 public class SqlServerDialectFactory implements JdbcDialectFactory {
     @Override
     public boolean acceptsURL(String url) {
-        return url.startsWith("jdbc:sqlserver:");
+        return (url.startsWith("jdbc:jtds:sqlserver:") || url.startsWith("jdbc:sqlserver:"));
     }
 
     @Override

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/sqlserver/SqlserverTypeMapper.java

@@ -57,6 +57,7 @@ public class SqlserverTypeMapper implements JdbcDialectTypeMapper {
     private static final String SQLSERVER_NTEXT = "NTEXT";
     private static final String SQLSERVER_NCHAR = "NCHAR";
     private static final String SQLSERVER_NVARCHAR = "NVARCHAR";
+    private static final String SQLSERVER_SYSNAME = "SYSNAME";
     private static final String SQLSERVER_TEXT = "TEXT";
 
     // ------------------------------time-------------------------
@@ -106,6 +107,7 @@ public SeaTunnelDataType<?> mapping(ResultSetMetaData metadata, int colIndex)
             case SQLSERVER_NTEXT:
             case SQLSERVER_NVARCHAR:
             case SQLSERVER_TEXT:
+            case SQLSERVER_SYSNAME:
                 return BasicType.STRING_TYPE;
             case SQLSERVER_DATE:
                 return LocalTimeType.LOCAL_DATE_TYPE;

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-3/pom.xml

@@ -91,6 +91,11 @@
             <artifactId>vertica-jdbc</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>net.sourceforge.jtds</groupId>
+            <artifactId>jtds</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-3/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcJtdsSqlServerIT.java

@@ -0,0 +1,160 @@
+/*
+ * 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.jdbc;
+
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+
+import org.apache.commons.lang3.tuple.Pair;
+
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.MSSQLServerContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import com.google.common.collect.Lists;
+
+import java.io.IOException;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class JdbcJtdsSqlServerIT extends AbstractJdbcIT {
+    private static final String SQLSERVER_IMAGE = "mcr.microsoft.com/mssql/server:2022-latest";
+    private static final String SQLSERVER_CONTAINER_HOST = "sqlserver";
+    private static final String SQLSERVER_SOURCE = "source";
+    private static final String SQLSERVER_SINK = "sink";
+    private static final int SQLSERVER_CONTAINER_PORT = 1433;
+    private static final String SQLSERVER_URL =
+            "jdbc:jtds:sqlserver://" + AbstractJdbcIT.HOST + ":%s";
+    private static final String DRIVER_CLASS = "net.sourceforge.jtds.jdbc.Driver";
+    private static final List<String> CONFIG_FILE =
+            Lists.newArrayList("/jdbc_jtds_sqlserver_source_to_sink.conf");
+    private static final String CREATE_SQL =
+            "CREATE TABLE %s (\n"
+                    + "  [age] bigint  NOT NULL,\n"
+                    + "  [name] varchar(255) COLLATE Chinese_PRC_CI_AS  NULL\n"
+                    + ")";
+
+    private String username;
+
+    private String password;
+
+    @Override
+    JdbcCase getJdbcCase() {
+        Map<String, String> containerEnv = new HashMap<>();
+        String jdbcUrl = String.format(SQLSERVER_URL, SQLSERVER_CONTAINER_PORT);
+        Pair<String[], List<SeaTunnelRow>> testDataSet = initTestData();
+        String[] fieldNames = testDataSet.getKey();
+
+        String insertSql = insertTable("", SQLSERVER_SOURCE, fieldNames);
+
+        return JdbcCase.builder()
+                .dockerImage(SQLSERVER_IMAGE)
+                .networkAliases(SQLSERVER_CONTAINER_HOST)
+                .containerEnv(containerEnv)
+                .driverClass(DRIVER_CLASS)
+                .host(AbstractJdbcIT.HOST)
+                .port(SQLSERVER_CONTAINER_PORT)
+                .localPort(SQLSERVER_CONTAINER_PORT)
+                .jdbcTemplate(SQLSERVER_URL)
+                .jdbcUrl(jdbcUrl)
+                .userName(username)
+                .password(password)
+                .sourceTable(SQLSERVER_SOURCE)
+                .sinkTable(SQLSERVER_SINK)
+                .createSql(CREATE_SQL)
+                .configFile(CONFIG_FILE)
+                .insertSql(insertSql)
+                .testData(testDataSet)
+                .build();
+    }
+
+    @Override
+    void compareResult() throws SQLException, IOException {}
+
+    @Override
+    String driverUrl() {
+        return "https://repo1.maven.org/maven2/net/sourceforge/jtds/jtds/1.3.1/jtds-1.3.1.jar";
+    }
+
+    @Override
+    Pair<String[], List<SeaTunnelRow>> initTestData() {
+        String[] fieldNames =
+                new String[] {
+                    "age", "name",
+                };
+
+        List<SeaTunnelRow> rows = new ArrayList<>();
+        for (int i = 0; i < 100; i++) {
+            SeaTunnelRow row =
+                    new SeaTunnelRow(
+                            new Object[] {
+                                i, "f_" + i,
+                            });
+            rows.add(row);
+        }
+
+        return Pair.of(fieldNames, rows);
+    }
+
+    @Override
+    GenericContainer<?> initContainer() {
+        DockerImageName imageName = DockerImageName.parse(SQLSERVER_IMAGE);
+
+        MSSQLServerContainer<?> container =
+                new MSSQLServerContainer<>(imageName)
+                        .withNetwork(TestSuiteBase.NETWORK)
+                        .withNetworkAliases(SQLSERVER_CONTAINER_HOST)
+                        .acceptLicense()
+                        .withLogConsumer(
+                                new Slf4jLogConsumer(
+                                        DockerLoggerFactory.getLogger(SQLSERVER_IMAGE)));
+
+        container.setPortBindings(
+                Lists.newArrayList(
+                        String.format(
+                                "%s:%s", SQLSERVER_CONTAINER_PORT, SQLSERVER_CONTAINER_PORT)));
+
+        try {
+            Class.forName(container.getDriverClassName());
+        } catch (ClassNotFoundException e) {
+            throw new SeaTunnelRuntimeException(
+                    JdbcITErrorCode.DRIVER_NOT_FOUND, "Not found suitable driver for mssql", e);
+        }
+
+        username = container.getUsername();
+        password = container.getPassword();
+
+        return container;
+    }
+
+    @Override
+    public String quoteIdentifier(String field) {
+        return "[" + field + "]";
+    }
+
+    @Override
+    public void clearTable(String schema, String table) {
+        // do nothing.
+    }
+}

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-3/src/test/resources/jdbc_jtds_sqlserver_source_to_sink.conf

@@ -0,0 +1,62 @@
+#
+# 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.
+#
+######
+###### This config file is a demonstration of streaming processing in seatunnel config
+######
+
+env {
+  # You can set flink configuration here
+  execution.parallelism = 1
+  job.mode = "BATCH"
+  #execution.checkpoint.interval = 10000
+  #execution.checkpoint.data-uri = "hdfs://localhost:9000/checkpoint"
+}
+
+source {
+  # This is a example source plugin **only for test and demonstrate the feature source plugin**
+  Jdbc {
+    # default port is 1433
+    driver = net.sourceforge.jtds.jdbc.Driver
+    url = "jdbc:jtds:sqlserver://sqlserver:port/database"
+    user = SA
+    password = "A_Str0ng_Required_Password"
+    query = "select age, name from source"
+  }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of source plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/source/Jdbc
+}
+
+transform {
+
+  # If you would like to get more information about how to configure seatunnel and see full list of transform plugins,
+  # please go to https://seatunnel.apache.org/docs/transform-v2/sql
+}
+
+sink {
+  Jdbc {
+    # default port is 1433
+    driver = net.sourceforge.jtds.jdbc.Driver
+    url = "jdbc:jtds:sqlserver://sqlserver:port/database"
+    user = SA
+    password = "A_Str0ng_Required_Password"
+    query = "insert into sink(age, name) values(?,?)"
+  }
+
+  # If you would like to get more information about how to configure seatunnel and see full list of sink plugins,
+  # please go to https://seatunnel.apache.org/docs/connector-v2/sink/Jdbc
+}