[Feature][Connector-V2] Add Apache InLong Connector Support #9250

luchunliang · 2025-04-29T04:05:54Z

Description

Background
Apache InLong is a one-stop data integration framework that supports efficient streaming/batch data collection, aggregation, and distribution from diverse sources.

Apache InLong primarily focuses on ultra-large-scale data integration but lacks support for diverse data sources. In contrast, SeaTunnel provides extensive data source compatibility. By adding an InLong Sink Connector to SeaTunnel, users can reuse SeaTunnel's rich data sources to feed data into InLong, bridging the gap between versatile data ingestion and large-scale integration capabilities, and gain seamless integration with the InLong ecosystem, enhancing real-time data pipeline capabilities for scenarios like log/metrics synchronization and multi-system data distribution.

Proposal
Implement an Apache InLong Connector for SeaTunnel with the following features:

InLong Sink Connector: Write processed data to InLong clusters for downstream consumption or governance.
Support core InLong configurations (e.g., groupId, streamId, authentication).

Module Structure

Create a new module: seatunnel-connectors/connector-inlong, referencing existing message queue connectors (e.g., Pulsar).
Leverage the InLong Java SDK (e.g., inlong-sdk-java) or direct API calls.

Usage Scenario

Distributing SeaTunnel-processed results through InLong to downstream systems (e.g., real-time dashboards, risk engines).
Suggested Solution

Purpose of this pull request

Add Apache InLong Connector Support

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 PR adds support for Apache InLong as a new sink connector in SeaTunnel, enabling users to write processed data into InLong clusters. Key changes include the implementation of core sink writer logic with retry and resend mechanisms, definition of state and commit objects, and integration with InLong’s Java SDK.

Added a new module for the InLong sink connector with complete sink writer, committer, and factory implementations.
Introduced utility methods and thread management for sending data with retry logic.
Provided unit tests and configuration properties for InLong connector functionality.

Reviewed Changes

Copilot reviewed 15 out of 17 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
seatunnel-connectors-v2/connector-inlong/src/test/java/org/apache/seatunnel/connectors/seatunnel/inlong/InlongFactoryTest.java	Basic unit tests for the InLongSinkFactory.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/util/InlongUtil.java	Utility class for sleep methods with logging.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/state/*.java	New state and commit info classes required for sink state management.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/SenderMessage.java	Defined the message container for batch sending.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongThreadFactory.java	New thread factory implementation for the connector’s internal threads.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java	Core sink writer implementation featuring message batching, retry, and resend logic.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkFactory.java	Factory to create the InLong sink and define required options.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkCommitter.java	Minimal committer implementation returning empty commit info.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSink.java	Main sink implementation integrating writer, state serializer, and committer.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/exception/*.java	New exception and error code classes for error handling in the connector.
seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/config/*.java	Configuration property definitions and semantics enums for the InLong sink connector.

Files not reviewed (2)

plugin-mapping.properties: Language not supported
seatunnel-connectors-v2/connector-inlong/pom.xml: Language not supported

Comments suppressed due to low confidence (2)

seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java:290

The current logic removes an extra element from rowQue when a message exceeds batchSendLen, which may unintentionally drop the subsequent message. Consider removing the redundant rowQue.remove() call so that only the oversized message is dropped.

if (peekMessageLength > batchSendLen) { LOG.warn(...); rowQue.remove(); break; }

seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java:247

In the retry loop for sending messages, the code does not break out after exceeding MAX_RETRY, which may lead to an infinite loop when sending continually fails. Consider adding a break or alternative error handling to prevent the potential deadlock.

if (retry > MAX_RETRY) { ... }

Hisoka-X · 2025-04-29T11:26:37Z

Thanks @luchunliang . Please add e2e test case.

Hisoka-X · 2025-05-15T11:19:28Z

Is it necessary to provide a MockServer to complete E2E test cases?

+1

luchunliang · 2025-05-28T03:33:23Z

Thanks @luchunliang . Please add e2e test case.

Offline, we have completed end-to-end testing. Here, by referencing connector-activemq and connector-aerospike, we have supplemented the test cases.

hailin0 · 2025-05-28T04:20:59Z

Thanks @luchunliang

add docs
https://github.com/apache/seatunnel/tree/dev/docs/en/connector-v2/sink
https://github.com/apache/seatunnel/tree/dev/docs/zh/connector-v2/sink
add testcase or e2e

ref
https://github.com/apache/seatunnel/tree/dev/seatunnel-e2e/seatunnel-connector-v2-e2e
https://github.com/apache/seatunnel/blob/dev/seatunnel-connectors-v2/connector-common/src/test/java/org/apache/seatunnel/connectors/seatunnel/sink/SinkFlowTestUtils.java

add into dist module pom.xml
https://github.com/apache/seatunnel/blob/dev/seatunnel-dist/pom.xml#L973
add ci label
https://github.com/apache/seatunnel/blob/dev/.github/workflows/labeler/label-scope-conf.yml#L314
enable your repo ci action

Copilot

Pull Request Overview

Add an Apache InLong Sink connector module to SeaTunnel, including configuration, factory, writer, committer, and registration.

Introduce connector-inlong module with core sink implementation and state classes
Add configuration options, semantics enum, and error handling for InLong SDK
Register connector in plugin mapping and provide a simple factory test

Reviewed Changes

Copilot reviewed 17 out of 17 changed files in this pull request and generated 6 comments.

Show a summary per file

File	Description
InlongFactoryTest.java	Add basic factory identifier tests
util/InlongUtil.java	Add sleep helper with exception handling
state/InlongSinkState.java	Define empty sink state class
state/InlongCommitInfo.java	Define commit info class
state/InlongAggregatedCommitInfo.java	Define aggregated commit info class
sink/SenderMessage.java	Represent batched messages for sending
sink/InlongThreadFactory.java	Create named threads with placeholder tags
sink/InlongSinkWriter.java	Implement async sink writer with retry logic
sink/InlongSinkFactory.java	Provide table sink factory
sink/InlongSinkCommitter.java	Define no-op committer
sink/InlongSink.java	Tie writer and committer into SeaTunnel API
exception/InlongException.java	Wrap SeaTunnel runtime exceptions
exception/InlongErrorCode.java	Enumerate connector-specific error codes
config/SinkProperties.java	Define connector configuration options
config/InlongSemantics.java	Define delivery semantics enum
connector-inlong/pom.xml	Add module dependencies and metadata
plugin-mapping.properties	Register `seatunnel.sink.Inlong` connector

Comments suppressed due to low confidence (5)

seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/config/InlongSemantics.java:26

The enum value NON is ambiguous; consider renaming to NONE or NO_SEMANTICS for clarity.

    NON,

seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/config/SinkProperties.java:50

Update the description to reference InLong instead of Pulsar to match this connector.

                    .withDescription("Configure which fields are used as the key of the pulsar message.");

seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongThreadFactory.java:33

[nitpick] Avoid Hungarian prefix m; rename mThreadNum to threadCounter or similar for consistency.

    private final AtomicInteger mThreadNum = new AtomicInteger(1);

seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java:89

[nitpick] Field EXECUTOR_SERVICE is mutable but named like a constant; consider lowercase or make it final.

    private static ThreadPoolExecutor EXECUTOR_SERVICE =

seatunnel-connectors-v2/connector-inlong/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java:225

[nitpick] Abbreviated variable suc is unclear; rename to success for readability.

        boolean suc = false;

Copilot · 2025-05-29T02:22:08Z

...g/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/config/InlongSemantics.java

+     * At this semantics, we will directly send the message to pulsar, the data may duplicat/lost if
+     * job restart/retry or network error.


Fix typo and improve grammar: use 'duplicated or lost' instead of 'duplicat/lost', and consider 'At this semantic'.

Suggested change

* At this semantics, we will directly send the message to pulsar, the data may duplicat/lost if

* job restart/retry or network error.

* In this semantic, we will directly send the message to Pulsar. The data may be duplicated or lost if

* the job restarts/retries or encounters a network error.

Copilot · 2025-05-29T02:22:09Z

...ng/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java

+                }
+                InlongUtil.silenceSleepInMs(BATCH_SEND_INTERVAL);
+            }
+            LOG.info("start send {}:{}", groupId, streamId);


The final log incorrectly repeats 'start send'; change it to 'stop send' to reflect shutdown.

Suggested change

LOG.info("start send {}:{}", groupId, streamId);

LOG.info("stop send {}:{}", groupId, streamId);

Copilot · 2025-05-29T02:22:09Z

...ng/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java

+                        "message size is {}, greater than max pack size {}, drop it!",
+                        peekMessage.length,
+                        batchSendLen);
+                rowQue.remove();


This extra remove() call can drop an unintended message. Only remove once when discarding oversized entries.

Suggested change

rowQue.remove();

Copilot · 2025-05-29T02:22:09Z

...ng/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java

+    public static final int MAX_RETRY = 5;
+    public static final int ERROR_LOG_SAMPLE = 10;
+    public static final int RESEND_QUEUE_WAIT_MS = 10;
+    private final long WRITE_INTERVAL_MS = 10;


[nitpick] Constant timing values should be declared static final to avoid per-instance allocation.

Suggested change

private final long WRITE_INTERVAL_MS = 10;

private static final long WRITE_INTERVAL_MS = 10;

Copilot · 2025-05-29T02:22:09Z

...ng/src/main/java/org/apache/seatunnel/connectors/seatunnel/inlong/sink/InlongSinkWriter.java

+        private final int msgCnt;
+
+        SenderCallback(SenderMessage message, int retry) {
+            this.message = message;
+            this.retry = retry;
+            this.msgCnt = message.getDataList().size();


Field msgCnt is never used; consider removing it to reduce dead code.

Suggested change

private final int msgCnt;

SenderCallback(SenderMessage message, int retry) {

this.message = message;

this.retry = retry;

this.msgCnt = message.getDataList().size();

// Removed unused field `msgCnt`.

SenderCallback(SenderMessage message, int retry) {

this.message = message;

this.retry = retry;

// Removed initialization of unused field `msgCnt`.

Copilot · 2025-05-29T02:22:10Z

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

+    </parent>
+
+    <artifactId>connector-inlong</artifactId>
+    <name>SeaTunnel : Connectors V2 : Pulsar</name>


Update the module <name> to reflect InLong instead of Pulsar.

hailin0 · 2025-05-29T02:56:21Z

add this connector into plugin_config
https://github.com/apache/seatunnel/blob/dev/config/plugin_config

github-actions bot added the connectors-v2 label Apr 29, 2025

nielifeng requested a review from Copilot April 29, 2025 06:28

Copilot AI reviewed Apr 29, 2025

View reviewed changes

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

[Feature][Connector-V2] Add Apache InLong Connector Support #9249

Closed

3 tasks

github-actions bot added document CI&CD e2e labels May 12, 2025

github-actions bot removed document CI&CD e2e labels May 28, 2025

[Feature][Connector-V2] Add Apache InLong Connector Support

ee50fc1

luchunliang force-pushed the seatunnel-9249 branch from ab421d9 to ee50fc1 Compare May 28, 2025 02:58

nielifeng requested a review from Copilot May 29, 2025 02:19

Copilot AI reviewed May 29, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

[Feature][Connector-V2] Add Apache InLong Connector Support #9250

[Feature][Connector-V2] Add Apache InLong Connector Support #9250

luchunliang commented Apr 29, 2025 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Hisoka-X commented Apr 29, 2025

Uh oh!

Hisoka-X commented May 15, 2025

Uh oh!

luchunliang commented May 28, 2025

Uh oh!

hailin0 commented May 28, 2025 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI May 29, 2025

Uh oh!

Copilot AI May 29, 2025

Uh oh!

Copilot AI May 29, 2025

Uh oh!

Copilot AI May 29, 2025

Uh oh!

Copilot AI May 29, 2025

Uh oh!

Copilot AI May 29, 2025

Uh oh!

hailin0 commented May 29, 2025

Uh oh!

Uh oh!

		* At this semantics, we will directly send the message to pulsar, the data may duplicat/lost if
		* job restart/retry or network error.

	LOG.info("start send {}:{}", groupId, streamId);
	LOG.info("stop send {}:{}", groupId, streamId);

	private final long WRITE_INTERVAL_MS = 10;
	private static final long WRITE_INTERVAL_MS = 10;

[Feature][Connector-V2] Add Apache InLong Connector Support #9250

Are you sure you want to change the base?

[Feature][Connector-V2] Add Apache InLong Connector Support #9250

Conversation

luchunliang commented Apr 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Usage Scenario

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!

Hisoka-X commented Apr 29, 2025

Uh oh!

Hisoka-X commented May 15, 2025

Uh oh!

luchunliang commented May 28, 2025

Uh oh!

hailin0 commented May 28, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI May 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI May 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI May 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI May 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI May 29, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI May 29, 2025

Choose a reason for hiding this comment

Uh oh!

hailin0 commented May 29, 2025

Uh oh!

Uh oh!

luchunliang commented Apr 29, 2025 •

edited

Loading

hailin0 commented May 28, 2025 •

edited

Loading