[Feature][Conenctor-V2] Add openmldb source connector (#3313)

* [Feature][Connector-V2][OpenMldb] First commit

* [Feature][Connector-V2] Add openmldb source connector

* [Feature][Connector-V2] Fix code style

* [Feature][Connector-V2] Update plugin-mapping.properties

* [Feature][Connector-V2] Update pom of seatunnel-dist

* [Feature][Connector-V2] Fix some bugs

* [Feature][Connector-V2] Fix some bugs

* [Feature][Connector-V2] Fix some bugs

* [Feature][Connector-V2] Make code more clearly

* [Feature][Connector-V2] Add docs

* [Feature][Connector-V2][OpenMldb] Unified exception

* [Feature][Connector-V2][OpenMldb] Add option factory

* [Feature][Connector-V2][OpenMldb] Unified exception

* [Feature][Connector-V2][OpenMldb] Fix code style

* [Feature][Connector-V2][OpenMldb] Fix pom error

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

@@ -0,0 +1,85 @@
+# OpenMldb
+
+> OpenMldb source connector
+
+## Description
+
+Used to read data from OpenMldb.
+
+## Key features
+
+- [x] [batch](../../concept/connector-v2-features.md)
+- [x] [stream](../../concept/connector-v2-features.md)
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+- [ ] [schema projection](../../concept/connector-v2-features.md)
+- [ ] [parallelism](../../concept/connector-v2-features.md)
+- [ ] [support user-defined split](../../concept/connector-v2-features.md)
+
+## Options
+
+| name            | type    | required | default value |
+|-----------------|---------|----------|---------------|
+| cluster_mode    | boolean | yes      | -             |
+| sql             | string  | yes      | -             |
+| database        | string  | yes      | -             |
+| host            | string  | no       | -             |
+| port            | int     | no       | -             |
+| zk_path         | string  | no       | -             |
+| zk_host         | string  | no       | -             |
+| session_timeout | int     | no       | 10000         |
+| request_timeout | int     | no       | 60000         |
+| common-options  |         | no       | -             |
+
+### cluster_mode [string]
+
+OpenMldb is or not cluster mode
+
+### sql [string]
+
+Sql statement
+
+### database [string]
+
+Database name
+
+### host [string]
+
+OpenMldb host, only supported on OpenMldb single mode
+
+### port [int]
+
+OpenMldb port, only supported on OpenMldb single mode
+
+### zk_host [string]
+
+Zookeeper host, only supported on OpenMldb cluster mode
+
+### zk_path [string]
+
+Zookeeper path, only supported on OpenMldb cluster mode
+
+### session_timeout [int]
+
+OpenMldb session timeout(ms), default 60000
+
+### request_timeout [int]
+
+OpenMldb request timeout(ms), default 10000
+
+### common options
+
+Source plugin common parameters, please refer to [Source Common Options](common-options.md) for details
+
+## Example
+
+```hocon
+
+  OpenMldb {
+    host = "172.17.0.2"
+    port = 6527
+    sql = "select * from demo_table1"
+    database = "demo_db"
+    cluster_mode = false
+  }
+
+```

plugin-mapping.properties

@@ -155,4 +155,5 @@ seatunnel.source.OneSignal = connector-http-onesignal
 seatunnel.source.Jira = connector-http-jira
 seatunnel.source.Gitlab = connector-http-gitlab
 seatunnel.sink.RabbitMQ = connector-rabbitmq
-seatunnel.source.RabbitMQ = connector-rabbitmq
+seatunnel.source.RabbitMQ = connector-rabbitmq
+seatunnel.source.OpenMldb = connector-openmldb

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

@@ -0,0 +1,54 @@
+<?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>seatunnel-connectors-v2</artifactId>
+        <groupId>org.apache.seatunnel</groupId>
+        <version>${revision}</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>connector-openmldb</artifactId>
+
+    <properties>
+        <openmldb.version>0.6.3</openmldb.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.4paradigm.openmldb</groupId>
+            <artifactId>openmldb-jdbc</artifactId>
+            <version>${openmldb.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.4paradigm.openmldb</groupId>
+            <artifactId>openmldb-native</artifactId>
+            <version>${openmldb.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>

seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/config/OpenMldbConfig.java

@@ -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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.openmldb.config;
+
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+public class OpenMldbConfig {
+    private static final int DEFAULT_SESSION_TIMEOUT = 10000;
+    private static final int DEFAULT_REQUEST_TIMEOUT = 60000;
+    public static final Option<String> ZK_HOST = Options.key("zk_host")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Zookeeper server host");
+    public static final Option<String> ZK_PATH = Options.key("zk_path")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Zookeeper server path of OpenMldb cluster");
+    public static final Option<String> HOST = Options.key("host")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("OpenMldb host");
+    public static final Option<Integer> PORT = Options.key("port")
+            .intType()
+            .noDefaultValue()
+            .withDescription("OpenMldb port");
+    public static final Option<Integer> SESSION_TIMEOUT = Options.key("session_timeout")
+            .intType()
+            .defaultValue(DEFAULT_SESSION_TIMEOUT)
+            .withDescription("OpenMldb session timeout");
+    public static final Option<Integer> REQUEST_TIMEOUT = Options.key("request_timeout")
+            .intType()
+            .defaultValue(DEFAULT_REQUEST_TIMEOUT)
+            .withDescription("OpenMldb request timeout");
+    public static final Option<Boolean> CLUSTER_MODE = Options.key("cluster_mode")
+            .booleanType()
+            .noDefaultValue()
+            .withDescription("Whether cluster mode is enabled");
+    public static final Option<String> SQL = Options.key("sql")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("Sql statement");
+    public static final Option<String> DATABASE = Options.key("database")
+            .stringType()
+            .noDefaultValue()
+            .withDescription("The database you want to access");
+}

seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/config/OpenMldbParameters.java

@@ -0,0 +1,73 @@
+/*
+ * 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.openmldb.config;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import lombok.Getter;
+
+import java.io.Serializable;
+
+@Getter
+public class OpenMldbParameters implements Serializable {
+    private String zkHost;
+    private String zkPath;
+    private String host;
+    private int port;
+    private int sessionTimeout = OpenMldbConfig.SESSION_TIMEOUT.defaultValue();
+    private int requestTimeout = OpenMldbConfig.REQUEST_TIMEOUT.defaultValue();
+    private Boolean clusterMode;
+    private String database;
+    private String sql;
+
+    private OpenMldbParameters() {
+        // do nothing
+    }
+
+    public static OpenMldbParameters buildWithConfig(Config pluginConfig) {
+        OpenMldbParameters openMldbParameters = new OpenMldbParameters();
+        openMldbParameters.clusterMode = pluginConfig.getBoolean(OpenMldbConfig.CLUSTER_MODE.key());
+        openMldbParameters.database = pluginConfig.getString(OpenMldbConfig.DATABASE.key());
+        openMldbParameters.sql = pluginConfig.getString(OpenMldbConfig.SQL.key());
+        // set zkHost
+        if (pluginConfig.hasPath(OpenMldbConfig.ZK_HOST.key())) {
+            openMldbParameters.zkHost = pluginConfig.getString(OpenMldbConfig.ZK_HOST.key());
+        }
+        // set zkPath
+        if (pluginConfig.hasPath(OpenMldbConfig.ZK_PATH.key())) {
+            openMldbParameters.zkPath = pluginConfig.getString(OpenMldbConfig.ZK_PATH.key());
+        }
+        // set host
+        if (pluginConfig.hasPath(OpenMldbConfig.HOST.key())) {
+            openMldbParameters.host = pluginConfig.getString(OpenMldbConfig.HOST.key());
+        }
+        // set port
+        if (pluginConfig.hasPath(OpenMldbConfig.PORT.key())) {
+            openMldbParameters.port = pluginConfig.getInt(OpenMldbConfig.PORT.key());
+        }
+        // set session timeout
+        if (pluginConfig.hasPath(OpenMldbConfig.SESSION_TIMEOUT.key())) {
+            openMldbParameters.sessionTimeout = pluginConfig.getInt(OpenMldbConfig.SESSION_TIMEOUT.key());
+        }
+        // set request timeout
+        if (pluginConfig.hasPath(OpenMldbConfig.REQUEST_TIMEOUT.key())) {
+            openMldbParameters.requestTimeout = pluginConfig.getInt(OpenMldbConfig.REQUEST_TIMEOUT.key());
+        }
+        return openMldbParameters;
+    }
+}

seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/config/OpenMldbSqlExecutor.java

@@ -0,0 +1,66 @@
+/*
+ * 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.openmldb.config;
+
+import com._4paradigm.openmldb.sdk.SdkOption;
+import com._4paradigm.openmldb.sdk.SqlException;
+import com._4paradigm.openmldb.sdk.impl.SqlClusterExecutor;
+
+public class OpenMldbSqlExecutor {
+    private static final SdkOption SDK_OPTION = new SdkOption();
+    private static volatile SqlClusterExecutor SQL_EXECUTOR;
+
+    private OpenMldbSqlExecutor() {
+
+    }
+
+    public static void initSdkOption(OpenMldbParameters openMldbParameters) {
+        if (openMldbParameters.getClusterMode()) {
+            SDK_OPTION.setZkCluster(openMldbParameters.getZkHost());
+            SDK_OPTION.setZkPath(openMldbParameters.getZkPath());
+        } else {
+            SDK_OPTION.setHost(openMldbParameters.getHost());
+            SDK_OPTION.setPort(openMldbParameters.getPort());
+            SDK_OPTION.setClusterMode(false);
+        }
+        SDK_OPTION.setSessionTimeout(openMldbParameters.getSessionTimeout());
+        SDK_OPTION.setRequestTimeout(openMldbParameters.getRequestTimeout());
+    }
+
+    public static SqlClusterExecutor getSqlExecutor() throws SqlException {
+        if (SQL_EXECUTOR == null) {
+            synchronized (OpenMldbSqlExecutor.class) {
+                if (SQL_EXECUTOR == null) {
+                    SQL_EXECUTOR = new SqlClusterExecutor(SDK_OPTION);
+                }
+            }
+        }
+        return SQL_EXECUTOR;
+    }
+
+    public static void close() {
+        if (SQL_EXECUTOR != null) {
+            synchronized (OpenMldbParameters.class) {
+                if (SQL_EXECUTOR != null) {
+                    SQL_EXECUTOR.close();
+                    SQL_EXECUTOR = null;
+                }
+            }
+        }
+    }
+}

seatunnel-connectors-v2/connector-openmldb/src/main/java/org/apache/seatunnel/connectors/seatunnel/openmldb/exception/OpenMldbConnectorException.java

@@ -0,0 +1,35 @@
+/*
+ * 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.openmldb.exception;
+
+import org.apache.seatunnel.common.exception.SeaTunnelErrorCode;
+import org.apache.seatunnel.common.exception.SeaTunnelRuntimeException;
+
+public class OpenMldbConnectorException extends SeaTunnelRuntimeException {
+    public OpenMldbConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage) {
+        super(seaTunnelErrorCode, errorMessage);
+    }
+
+    public OpenMldbConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, String errorMessage, Throwable cause) {
+        super(seaTunnelErrorCode, errorMessage, cause);
+    }
+
+    public OpenMldbConnectorException(SeaTunnelErrorCode seaTunnelErrorCode, Throwable cause) {
+        super(seaTunnelErrorCode, cause);
+    }
+}