[Feature][http-Sink] Implementing http batch writes (#9292)

Author: ocean-zhc

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

@@ -44,6 +44,9 @@ They can be downloaded via install-plugin.sh or from the Maven central repositor
 | retry_backoff_max_ms        | Int    | No       | 10000   | The maximum retry-backoff times(millis) if request http failed                                              |
 | connect_timeout_ms          | Int    | No       | 12000   | Connection timeout setting, default 12s.                                                                    |
 | socket_timeout_ms           | Int    | No       | 60000   | Socket timeout setting, default 60s.                                                                        |
+| array_mode                  | Boolean| No       | false   | Send data as a JSON array when true, or as a single JSON object when false (default)                        |
+| batch_size                  | Int    | No       | 1       | The batch size of records to send in one HTTP request. Only works when array_mode is true.                  |
+| request_interval_ms         | Int    | No       | 0       | The interval milliseconds between two HTTP requests, to avoid sending requests too frequently.              |
 | common-options              |        | No       | -       | Sink plugin common parameters, please refer to [Sink Common Options](../sink-common-options.md) for details |
 
 ## Example
@@ -59,6 +62,21 @@ Http {
 }
 ```
 
+### With Batch Processing
+
+```hocon
+Http {
+    url = "http://localhost/test/webhook"
+    headers {
+        token = "9e32e859ef044462a257e1fc76730066"
+        Content-Type = "application/json"
+    }
+    array_mode = true
+    batch_size = 50
+    request_interval_ms = 500
+}
+```
+
 ### Multiple table
 
 #### example1

docs/zh/connector-v2/sink/Http.md

@@ -42,6 +42,9 @@ import ChangeLog from '../changelog/connector-http.md';
 | retry_backoff_max_ms        | Int    | 否    | 10000 | http请求失败，最大重试回退时间(毫秒)                                      |
 | connect_timeout_ms          | Int    | 否    | 12000 | 连接超时设置，默认12s                                               |
 | socket_timeout_ms           | Int    | 否    | 60000 | 套接字超时设置，默认为60s                                             |
+| array_mode                  | Boolean| 否    | false | 为true时将数据作为JSON数组发送，为false时作为单个JSON对象发送（默认）                |
+| batch_size                  | Int    | 否    | 1     | 在一个HTTP请求中发送的记录批量大小。仅在array_mode为true时有效                   |
+| request_interval_ms         | Int    | 否    | 0     | 两次HTTP请求之间的间隔毫秒数，以避免请求过于频繁                                 |
 | common-options              |        | 否    | -     | Sink插件常用参数，请参考 [Sink常用选项 ](../sink-common-options.md) 了解详情 |
 
 ## 示例
@@ -57,6 +60,21 @@ Http {
 }
 ```
 
+### 带批处理的示例
+
+```hocon
+Http {
+    url = "http://localhost/test/webhook"
+    headers {
+        token = "9e32e859ef044462a257e1fc76730066"
+        Content-Type = "application/json"
+    }
+    array_mode = true
+    batch_size = 50
+    request_interval_ms = 500
+}
+```
+
 ## 变更日志
 
 <ChangeLog />

seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpParameter.java

@@ -42,6 +42,9 @@ public class HttpParameter implements Serializable {
     protected boolean enableMultilines;
     protected int connectTimeoutMs;
     protected int socketTimeoutMs;
+    protected boolean arrayMode = false;
+    protected int batchSize = 1;
+    protected int requestIntervalMs = 0;
 
     public void buildWithConfig(ReadonlyConfig pluginConfig) {
         // set url

seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSinkOptions.java

@@ -17,4 +17,27 @@
 
 package org.apache.seatunnel.connectors.seatunnel.http.config;
 
-public class HttpSinkOptions extends HttpCommonOptions {}
+import org.apache.seatunnel.api.configuration.Option;
+import org.apache.seatunnel.api.configuration.Options;
+
+public class HttpSinkOptions extends HttpCommonOptions {
+    public static final Option<Boolean> ARRAY_MODE =
+            Options.key("array_mode")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withDescription(
+                            "Send data as a JSON array when true, or as a single JSON object when false (default)");
+
+    public static final Option<Integer> BATCH_SIZE =
+            Options.key("batch_size")
+                    .intType()
+                    .defaultValue(1)
+                    .withDescription(
+                            "The batch size of records to send in one HTTP request. Only works when array_mode is true");
+
+    public static final Option<Integer> REQUEST_INTERVAL_MS =
+            Options.key("request_interval_ms")
+                    .intType()
+                    .defaultValue(0)
+                    .withDescription("The interval milliseconds between two HTTP requests");
+}

seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSink.java

@@ -47,6 +47,16 @@ public HttpSink(ReadonlyConfig pluginConfig, CatalogTable catalogTable) {
         if (pluginConfig.getOptional(HttpSinkOptions.PARAMS).isPresent()) {
             httpParameter.setHeaders(pluginConfig.get(HttpSinkOptions.PARAMS));
         }
+        if (pluginConfig.getOptional(HttpSinkOptions.ARRAY_MODE).isPresent()) {
+            httpParameter.setArrayMode(pluginConfig.get(HttpSinkOptions.ARRAY_MODE));
+        }
+        if (pluginConfig.getOptional(HttpSinkOptions.BATCH_SIZE).isPresent()) {
+            httpParameter.setBatchSize(pluginConfig.get(HttpSinkOptions.BATCH_SIZE));
+        }
+        if (pluginConfig.getOptional(HttpSinkOptions.REQUEST_INTERVAL_MS).isPresent()) {
+            httpParameter.setRequestIntervalMs(
+                    pluginConfig.get(HttpSinkOptions.REQUEST_INTERVAL_MS));
+        }
         this.catalogTable = catalogTable;
         this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();
     }

seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSinkFactory.java

@@ -48,6 +48,9 @@ public OptionRule optionRule() {
                 .optional(HttpSinkOptions.RETRY)
                 .optional(HttpSinkOptions.RETRY_BACKOFF_MULTIPLIER_MS)
                 .optional(HttpSinkOptions.RETRY_BACKOFF_MAX_MS)
+                .optional(HttpSinkOptions.ARRAY_MODE)
+                .optional(HttpSinkOptions.BATCH_SIZE)
+                .optional(HttpSinkOptions.REQUEST_INTERVAL_MS)
                 .optional(SinkConnectorCommonOptions.MULTI_TABLE_SINK_REPLICA)
                 .build();
     }

seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSinkWriter.java

@@ -17,6 +17,10 @@
 
 package org.apache.seatunnel.connectors.seatunnel.http.sink;
 
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.seatunnel.shade.com.fasterxml.jackson.databind.node.ArrayNode;
+import org.apache.seatunnel.shade.com.google.common.annotations.VisibleForTesting;
+
 import org.apache.seatunnel.api.serialization.SerializationSchema;
 import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -30,7 +34,10 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
 
 @Slf4j
 public class HttpSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
@@ -40,6 +47,13 @@ public class HttpSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
     protected final HttpParameter httpParameter;
     protected final SerializationSchema serializationSchema;
 
+    // Batch related fields
+    private final boolean arrayMode;
+    private final int batchSize;
+    private final int requestIntervalMs;
+    private final List<SeaTunnelRow> batchBuffer;
+    private long lastRequestTime;
+
     public HttpSinkWriter(SeaTunnelRowType seaTunnelRowType, HttpParameter httpParameter) {
         this(seaTunnelRowType, httpParameter, new JsonSerializationSchema(seaTunnelRowType));
     }
@@ -48,18 +62,81 @@ public HttpSinkWriter(
             SeaTunnelRowType seaTunnelRowType,
             HttpParameter httpParameter,
             SerializationSchema serializationSchema) {
+        this(
+                seaTunnelRowType,
+                httpParameter,
+                serializationSchema,
+                httpParameter.isArrayMode(),
+                httpParameter.getBatchSize(),
+                httpParameter.getRequestIntervalMs());
+    }
+
+    public HttpSinkWriter(
+            SeaTunnelRowType seaTunnelRowType,
+            HttpParameter httpParameter,
+            SerializationSchema serializationSchema,
+            boolean arrayMode,
+            int batchSize,
+            int requestIntervalMs) {
         this.seaTunnelRowType = seaTunnelRowType;
         this.httpParameter = httpParameter;
-        this.httpClient = new HttpClientProvider(httpParameter);
+        this.httpClient = createHttpClient(httpParameter);
         this.serializationSchema = serializationSchema;
+        this.arrayMode = arrayMode;
+        this.batchSize = batchSize;
+        this.requestIntervalMs = requestIntervalMs;
+        this.batchBuffer = new ArrayList<>(batchSize);
+        this.lastRequestTime = System.currentTimeMillis();
     }
 
     @Override
     public void write(SeaTunnelRow element) throws IOException {
+        if (!arrayMode) {
+            writeSingleRecord(element);
+        } else {
+            batchBuffer.add(element);
+            if (batchBuffer.size() >= batchSize) {
+                flush();
+            }
+        }
+    }
+
+    private void writeSingleRecord(SeaTunnelRow element) throws IOException {
         byte[] serialize = serializationSchema.serialize(element);
         String body = new String(serialize);
+        doHttpRequest(body);
+    }
+
+    private void flush() throws IOException {
+        if (batchBuffer.isEmpty()) {
+            return;
+        }
+        long currentTime = System.currentTimeMillis();
+        long timeSinceLastRequest = currentTime - lastRequestTime;
+        if (requestIntervalMs > 0 && timeSinceLastRequest < requestIntervalMs) {
+            try {
+                Thread.sleep(requestIntervalMs - timeSinceLastRequest);
+            } catch (InterruptedException e) {
+                throw new RuntimeException(e);
+            }
+        }
+
+        // Array mode: serialize batch data as JSON
+        ObjectMapper mapper = new ObjectMapper();
+        ArrayNode arrayNode = mapper.createArrayNode();
+        for (SeaTunnelRow row : batchBuffer) {
+            byte[] serialize = serializationSchema.serialize(row);
+            arrayNode.add(new String(serialize));
+        }
+        String body = mapper.writeValueAsString(arrayNode);
+        doHttpRequest(body);
+
+        batchBuffer.clear();
+        lastRequestTime = System.currentTimeMillis();
+    }
+
+    private void doHttpRequest(String body) {
         try {
-            // only support post web hook
             HttpResponse response =
                     httpClient.doPost(httpParameter.getUrl(), httpParameter.getHeaders(), body);
             if (HttpResponse.STATUS_OK == response.getCode()) {
@@ -76,8 +153,28 @@ public void write(SeaTunnelRow element) throws IOException {
 
     @Override
     public void close() throws IOException {
+        if (arrayMode) {
+            flush();
+        }
         if (Objects.nonNull(httpClient)) {
             httpClient.close();
         }
     }
+
+    @Override
+    public Optional<Void> prepareCommit() {
+        if (arrayMode) {
+            try {
+                flush();
+            } catch (IOException e) {
+                throw new RuntimeException("Failed to flush data in prepareCommit", e);
+            }
+        }
+        return Optional.empty();
+    }
+
+    @VisibleForTesting
+    protected HttpClientProvider createHttpClient(HttpParameter httpParameter) {
+        return new HttpClientProvider(httpParameter);
+    }
 }