[Feature][Connector-V2][Doris] Add Doris StreamLoad sink connector

docs/en/connector-v2/Error-Quick-Reference-Manual.md

@@ -159,6 +159,12 @@ problems encountered by users.
 | HUDI-01 | Create ParquetMetadata failed | When the user encounters this error code, it indicates that ParquetMetadata creation failed. Please check |
 | HUDI-02 | Kerberos Authorized failed    | When the user encounters this error code, it indicates that Kerberos authorization failed. Please check   |
 
+## Doris Connector Error Codes
+
+| code     | description                               | solution                                                                                                                                 |
+|----------|-------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------|
+| Doris-01 | Writing records to Doris failed.          | When users encounter this error code, it means that writing records to Doris failed, please check data from files whether is correct     |
+
 ## Clickhouse Connector Error Codes
 
 | code          | description                                                               | solution                                                                                                                                                                |

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

@@ -0,0 +1,130 @@
+# Doris
+
+> Doris sink connector
+
+## Description
+Used to send data to Doris. Both support streaming and batch mode.
+The internal implementation of Doris sink connector is cached and imported by stream load in batches.
+## Key features
+
+- [ ] [exactly-once](../../concept/connector-v2-features.md)
+- [ ] [schema projection](../../concept/connector-v2-features.md)
+
+## Options
+
+| name                        | type                         | required | default value   |
+|-----------------------------|------------------------------|----------|-----------------|
+| node_urls                   | list                         | yes      | -               |
+| username                    | string                       | yes      | -               |
+| password                    | string                       | yes      | -               |
+| database                    | string                       | yes      | -               |
+| table                       | string                       | yes      | -               |
+| labelPrefix                 | string                       | no       | -               |
+| batch_max_rows              | long                         | no       | 1024            |
+| batch_max_bytes             | int                          | no       | 5 * 1024 * 1024 |
+| batch_interval_ms           | int                          | no       | -               |
+| max_retries                 | int                          | no       | -               |
+| retry_backoff_multiplier_ms | int                          | no       | -               |
+| max_retry_backoff_ms        | int                          | no       | -               |
+| sink.properties.*           | doris stream load config     | no       | -               |
+
+### node_urls [list]
+
+`Doris` cluster address, the format is `["fe_ip:fe_http_port", ...]`
+
+### username [string]
+
+`Doris` user username
+
+### password [string]
+
+`Doris` user password
+
+### database [string]
+
+The name of `Doris` database
+
+### table [string]
+
+The name of `Doris` table
+
+### labelPrefix [string]
+
+The prefix of `Doris` stream load label
+
+### batch_max_rows [long]
+
+For batch writing, when the number of buffers reaches the number of `batch_max_rows` or the byte size of `batch_max_bytes` or the time reaches `batch_interval_ms`, the data will be flushed into the Doris
+
+### batch_max_bytes [int]
+
+For batch writing, when the number of buffers reaches the number of `batch_max_rows` or the byte size of `batch_max_bytes` or the time reaches `batch_interval_ms`, the data will be flushed into the Doris
+
+### batch_interval_ms [int]
+
+For batch writing, when the number of buffers reaches the number of `batch_max_rows` or the byte size of `batch_max_bytes` or the time reaches `batch_interval_ms`, the data will be flushed into the Doris
+
+### max_retries [int]
+
+The number of retries to flush failed
+
+### retry_backoff_multiplier_ms [int]
+
+Using as a multiplier for generating the next delay for backoff
+
+### max_retry_backoff_ms [int]
+
+The amount of time to wait before attempting to retry a request to `Doris`
+
+### sink.properties.*  [doris stream load config]
+
+The parameter of the stream load `data_desc`
+The way to specify the parameter is to add the prefix `sink.properties.` to the original stream load parameter name 
+For example, the way to specify `strip_outer_array` is: `sink.properties.strip_outer_array`
+
+#### Supported import data formats
+
+The supported formats include CSV and JSON. Default value: CSV
+
+## Example
+
+Use JSON format to import data
+
+```
+sink {
+    Doris {
+        nodeUrls = ["e2e_dorisdb:8030"]
+        username = root
+        password = ""
+        database = "test"
+        table = "e2e_table_sink"
+        batch_max_rows = 100
+        sink.properties.format = "JSON"
+        sink.properties.strip_outer_array = true
+    }
+}
+
+```
+
+Use CSV format to import data
+
+```
+sink {
+    Doris {
+        nodeUrls = ["e2e_dorisdb:8030"]
+        username = root
+        password = ""
+        database = "test"
+        table = "e2e_table_sink"
+        batch_max_rows = 100
+        sink.properties.format = "CSV"
+        sink.properties.column_separator = ","
+    }
+}
+```
+
+## Changelog
+
+### next version
+
+- Add Doris Sink Connector

plugin-mapping.properties

@@ -156,4 +156,5 @@ seatunnel.source.Jira = connector-http-jira
 seatunnel.source.Gitlab = connector-http-gitlab
 seatunnel.sink.RabbitMQ = connector-rabbitmq
 seatunnel.source.RabbitMQ = connector-rabbitmq
-seatunnel.source.OpenMldb = connector-openmldb
+seatunnel.source.OpenMldb = connector-openmldb
+seatunnel.sink.Doris = connector-doris

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

@@ -0,0 +1,68 @@
+<?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/maven-v4_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-doris</artifactId>
+
+    <properties>
+        <httpclient.version>4.5.13</httpclient.version>
+        <httpcore.version>4.4.4</httpcore.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>connector-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>${httpclient.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpcore</artifactId>
+            <version>${httpcore.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-json</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.seatunnel</groupId>
+            <artifactId>seatunnel-format-text</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+</project>

seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/client/DorisFlushTuple.java

@@ -0,0 +1,33 @@
+/*
+ * 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.doris.client;
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.List;
+
+@AllArgsConstructor
+@Getter
+@Setter
+public class DorisFlushTuple {
+    private String label;
+    private Long bytes;
+    private List<byte[]> rows;
+}