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

ocean-zhc · 2025-05-08T05:09:25Z

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

If any new Jar binary package adding in your PR, please add License Notice according
New License Guide
If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
If you are contributing the connector code, please check that the following files are updated:
1. Update plugin-mapping.properties and add new connector information in it
2. Update the pom file of seatunnel-dist
3. Add ci label in label-scope-conf
4. Add e2e testcase in seatunnel-e2e
5. Update connector plugin_config
Update the release-note.

Copilot

Pull Request Overview

This pull request implements HTTP batch writes for the HTTP sink connector in SeaTunnel. Key changes include:

Adding batch processing logic in HttpSinkWriter for both array and object modes
Introducing new configuration options (array_mode, batch_size, request_interval_ms, and format) for HTTP sinks
Updating both tests and documentation to verify and explain the new batch features

Reviewed Changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
seatunnel-connectors-v2/connector-http/connector-http-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSinkBatchWriterTest.java	Added tests for both object and array modes to validate individual and batch processing
seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSinkWriter.java	Enhanced writer logic with batch buffering and configurable HTTP request intervals in array mode
seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSinkFactory.java	Integrated new options into the sink factory for proper instantiation of the sink writer
seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/sink/HttpSink.java	Updated to invoke the new writer with batch processing parameters
seatunnel-connectors-v2/connector-http/connector-http-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSinkOptions.java	Added new options to support batch processing configuration
docs/zh/connector-v2/sink/Http.md docs/en/connector-v2/sink/Http.md	Updated documentation to describe the new batch processing features and their configuration

Copilot · 2025-05-09T01:42:18Z

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

+                byte[] serialize = serializationSchema.serialize(row);
+                jsonRecords.add(new String(serialize));
+            }
+            String body = "[" + String.join(",", jsonRecords) + "]";


[nitpick] Consider using a dedicated JSON library to construct the JSON array, which can help ensure proper formatting and handle edge cases more robustly.

The current way of manually constructing JSON arrays in the code does have some potential problems, I have constructed json arrays using Jackson

Copilot · 2025-05-09T01:42:18Z

...ase/src/main/java/org/apache/seatunnel/connectors/seatunnel/http/config/HttpSinkOptions.java

+                    .defaultValue(0)
+                    .withDescription("The interval milliseconds between two HTTP requests");
+
+    public static final Option<String> FORMAT =


[nitpick] The description for the FORMAT option indicates that only 'json' is supported; consider clarifying the documentation or renaming the option to avoid confusion regarding supported formats.

The Http.md documentation already states that

Hisoka-X · 2025-05-09T03:04:34Z

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

+| 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.              |
+| format                      | String | No       | json    | The format of batch data. Currently only "json" is supported, which will send data as JSON array.           |


what another format we can support in the future?

Currently think of just json, later expansion, the code can be modified to enumerate the judgment can be

Please remove useless config.

+1 removed useless config.

The docs should be updated too.

Hisoka-X · 2025-05-15T12:21:06Z

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

+| 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.              |
+| format                      | String | No       | json    | The format of batch data. Currently only "json" is supported, which will send data as JSON array.           |


The docs should be updated too.

Hisoka-X · 2025-05-15T12:22:42Z

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

+        boolean arrayMode = pluginConfig.get(HttpSinkOptions.ARRAY_MODE);
+        int batchSize = pluginConfig.get(HttpSinkOptions.BATCH_SIZE);
+        int requestIntervalMs = pluginConfig.get(HttpSinkOptions.REQUEST_INTERVAL_MS);


Please add those config into

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

Line 29 in 3783816

public class HttpParameter implements Serializable {

Hisoka-X · 2025-05-15T12:25:51Z

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

    }

    @Override
    public void write(SeaTunnelRow element) throws IOException {
+        if (!arrayMode) {
+            // Object mode: send each record individually, ignore batch_size setting


Suggested change

// Object mode: send each record individually, ignore batch_size setting

Hisoka-X · 2025-05-15T12:26:00Z

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

+            // Object mode: send each record individually, ignore batch_size setting
+            writeSingleRecord(element);
+        } else {
+            // Array mode: batch processing


Suggested change

// Array mode: batch processing

Hisoka-X · 2025-05-15T12:26:55Z

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

        try {
-            // only support post web hook
+            // Send HTTP request


Suggested change

// Send HTTP request

Hisoka-X · 2025-05-15T12:28:15Z

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

@@ -76,8 +166,15 @@ public void write(SeaTunnelRow element) throws IOException {

    @Override
    public void close() throws IOException {
+        if (arrayMode) {


we should invoke flush method when invoke prepareCommit method too.

Hisoka-X · 2025-05-15T12:29:27Z

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

        if (Objects.nonNull(httpClient)) {
            httpClient.close();
        }
    }
+
+    protected HttpClientProvider createHttpClient(HttpParameter httpParameter) {


Suggested change

protected HttpClientProvider createHttpClient(HttpParameter httpParameter) {

@VisibleForTesting

protected HttpClientProvider createHttpClient(HttpParameter httpParameter) {

Hisoka-X · 2025-05-15T12:30:19Z

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

+            return;
+        }
+
+        // Check request interval


Suggested change

// Check request interval

Please do not add useless comment.

Hisoka-X · 2025-05-15T12:31:48Z

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

+                Thread.currentThread().interrupt();
+                log.warn("Sleep interrupted", e);


We should throw exception directly. Othewise the writer woule never be closed.

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

[Feature][http-Sink] Implementing http batch writes

d1a933e

github-actions bot added document connectors-v2 http labels May 8, 2025

ocean-zhc mentioned this pull request May 8, 2025

[Feature][Sink] When the Source is LocalFile and the Sink is HTTP, how to set the batch read size? #9289

Closed

3 tasks

ocean-zhc added 3 commits May 8, 2025 17:27

Replace Chinese comments with English comments

54ccf2e

Replace Chinese comments with English comments

02d0fbd

Optimized code

5e74c48

hailin0 requested a review from Copilot May 9, 2025 01:41

Copilot AI reviewed May 9, 2025

View reviewed changes

ocean-zhc added 2 commits May 9, 2025 10:07

use jackson

25579f8

Fix package introduction issues with ObjectMapper and ArrayNode

c3758dc

Hisoka-X linked an issue May 9, 2025 that may be closed by this pull request

[Feature][Sink] When the Source is LocalFile and the Sink is HTTP, how to set the batch read size? #9289

Closed

3 tasks

Hisoka-X reviewed May 9, 2025

View reviewed changes

ocean-zhc and others added 6 commits May 9, 2025 11:41

Merge branch 'apache:dev' into httpsink-batch

f21c178

Merge branch 'apache:dev' into httpsink-batch

94c85b0

Merge branch 'apache:dev' into httpsink-batch

90162c5

Merge branch 'apache:dev' into httpsink-batch

1edda2c

Optimization Code

ee0bddd

Merge branch 'apache:dev' into httpsink-batch

a3ca1c9

Hisoka-X reviewed May 15, 2025

View reviewed changes

ocean-zhc and others added 6 commits May 19, 2025 09:17

Merge branch 'apache:dev' into httpsink-batch

2fde7fd

Optimization Code

14e7446

Optimization Code

f61cd41

Optimization Code

59a7d66

Optimization Code

5829b5e

Merge branch 'apache:dev' into httpsink-batch

dd180ca

Hisoka-X reviewed May 19, 2025

View reviewed changes

docs/zh/connector-v2/sink/Http.md Outdated Show resolved Hide resolved

docs/zh/connector-v2/sink/Http.md Outdated Show resolved Hide resolved

Hisoka-X added 2 commits May 19, 2025 20:21

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

a15aecc

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

68abc4c

Hisoka-X approved these changes May 19, 2025

View reviewed changes

github-actions bot added approved reviewed labels May 19, 2025

corgy-w approved these changes May 20, 2025

View reviewed changes

corgy-w merged commit 04ee8ac into apache:dev May 20, 2025
5 checks passed

ocean-zhc deleted the httpsink-batch branch May 20, 2025 13:45

joexjx pushed a commit to joexjx/seatunnel that referenced this pull request May 21, 2025

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

c71368d

	protected HttpClientProvider createHttpClient(HttpParameter httpParameter) {
	@VisibleForTesting
	protected HttpClientProvider createHttpClient(HttpParameter httpParameter) {

		Thread.currentThread().interrupt();
		log.warn("Sleep interrupted", e);

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

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

Uh oh!

Conversation

ocean-zhc commented May 8, 2025

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI May 9, 2025

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI May 9, 2025

Choose a reason for hiding this comment

Uh oh!

ocean-zhc May 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

ocean-zhc May 9, 2025 •

edited

Loading