[Feature] [Connector-V2] [IoTDB] Added source and sink connectors to support IoTDB 2.X by LJW21-02 · Pull Request #9872 · apache/seatunnel

LJW21-02 · 2025-09-18T02:41:17Z

Purpose of this pull request

This pull request adds both source and sink connectors to support IoTDB version >= 2.X.

Does this PR introduce any user-facing change?

No

How was this patch tested?

e2e tests

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

…DB-table

…leCompression in tableSession

Updated from seatunnel

zhangshenghang · 2025-09-18T23:51:23Z

+
+| 数据源   | 支持的版本                        | 地址             |
+|-------|------------------------------|----------------|
+| IoTDB | `0.13.0 <= version <= 1.3.X` | localhost:6667 |


This is incorrect

zhangshenghang · 2025-09-19T00:22:23Z

+        sinkConfig.setEnableRPCCompression(
+                pluginConfig.get(IoTDBSinkOptions.ENABLE_RPC_COMPRESSION));
+        if (pluginConfig.getOptional(IoTDBSinkOptions.CONNECTION_TIMEOUT_IN_MS).isPresent()) {
+            checkNotNull(sinkConfig.getEnableRPCCompression());


Why check here?

zhangshenghang · 2025-09-19T00:32:40Z

+
+        try {
+            if (session != null) {
+                session.close();


flush method exception, which can result in the inability to shut down resources

LJW21-02 · 2025-09-19T01:40:35Z

@zhangshenghang Hi, thank you for your review and suggestions! I’ve addressed the changes. Could you please take another look?

davidzollo · 2025-09-19T06:21:04Z

+                } else {
+                    session.insertRecords(
+                            batchRecords.getDeviceIds(),
+                            batchRecords.getTimestamps(),


Should there be a break in this for clause?

davidzollo · 2025-09-19T06:31:25Z

Please paste the result picture of your local running to check the accuracy. For example, using this task config file seatunnel-e2e/seatunnel-connector-v2-e2e/connector-iotdb-v2-e2e/src/test/resources/iotdb/iotdb_source_to_sink_table.conf, then you can check whether the count is the same between the sink side and the source side. You can use a small dataset to check.

LJW21-02 · 2025-09-24T01:20:33Z

@zhangshenghang Hi! All the checks have passed.

davidzollo · 2025-09-24T13:20:54Z

+| thrift_default_buffer_size | int     | No       | -             | The thrift_default_buffer_size of the IoTDB when you select                                                       |
+| thrift_max_frame_size      | int     | No       | -             | The thrift max frame size                                                                                         |
+| enable_cache_leader        | boolean | No       | -             | Enable_cache_leader of the IoTDB when you select                                                                  |
+| version                    | string  | No       | -             | SQL semantic version used by the client, The possible values are: `V_0_12`, `V_0_13`                              |


Does V2 still works fine in V_0_12, V_0_13? Please check

zhangshenghang · 2025-09-25T00:48:13Z

+
+> IoTDB 数据接收器
+
+## 描述


Please describe the supported engine.

LJW21-02 · 2025-09-25T10:42:54Z

@davidzollo @zhangshenghang Hi! Thank you for the feedback! I've implemented all the suggested changes and passed all the checks. Please review when you have a moment.

Hisoka-X · 2025-09-26T14:37:36Z

I am a little confused. Why is iotdb v2 a separate connector? Can't the original connector directly upgrade its dependencies to be compatible with v2?

davidzollo · 2025-09-27T08:09:42Z

+        for (int reader : readers) {
+            List<IoTDBv2SourceSplit> assignmentForReader = pendingSplit.remove(reader);
+            if (assignmentForReader != null && !assignmentForReader.isEmpty()) {
+                log.info("Assign splits {} to reader {}", assignmentForReader, reader);
+                try {
+                    context.assignSplit(reader, assignmentForReader);
+                } catch (Exception e) {
+                    log.error(
+                            "Failed to assign splits {} to reader {}",
+                            assignmentForReader,
+                            reader,
+                            e);
+                    pendingSplit.put(reader, assignmentForReader);
+                }
+            }
+        }


synchronized (stateLock) {
...
}

In this enumerator class, please guard every access to pendingSplit/shouldEnumerate with the same stateLock—wrap addPendingSplit, assignSplit, addSplitsBack, registerReader, and currentUnassignedSplitSize in that lock

davidzollo · 2025-09-27T08:09:46Z

+        int readerCount = context.currentParallelism();
+        for (IoTDBv2SourceSplit split : splits) {
+            int ownerReader = getSplitOwner(split.splitId(), readerCount);
+            log.info("Assigning {} to {} reader.", split, ownerReader);
+            pendingSplit.computeIfAbsent(ownerReader, r -> new ArrayList<>()).add(split);
+        }


pendingSplit is a competitive resource and should be locked，
synchronized (stateLock) {
...
}

LJW21-02 · 2025-09-28T01:56:58Z

I am a little confused. Why is iotdb v2 a separate connector? Can't the original connector directly upgrade its dependencies to be compatible with v2?

Hi @Hisoka-X ! We use a separate connector because each of them supports different IoTDB version, which also requires different dependency to handle things differently.

zhangshenghang · 2025-09-28T12:26:22Z

+
+    private ITableSession buildTableSession(ReadonlyConfig conf) throws IoTDBConnectionException {
+        TableSessionBuilder sessionBuilder = new TableSessionBuilder().enableCompression(false);
+        String nodeUrlsString = conf.get(NODE_URLS);


It should not be String; it should be List

zhangshenghang · 2025-09-28T12:27:30Z

+
+| 名称                         | 类型      | 是否必填 | 默认值  | 描述                                                                               |
+|----------------------------|---------|------|------|----------------------------------------------------------------------------------|
+| node_urls                  | string  | 是    | -    | IoTDB 集群地址，格式为 `"host1:port"` 或 `"host1:port,host2:port"`                        |


It should not be String; it should be List, and all other related parts should be modified.

zhangshenghang · 2025-09-28T12:32:05Z

I am a little confused. Why is iotdb v2 a separate connector? Can't the original connector directly upgrade its dependencies to be compatible with v2?

Hi @Hisoka-X ! We use a separate connector because each of them supports different IoTDB version, which also requires different dependency to handle things differently.

While review the code, I found that a lot of the code is duplicated with the existing code. Can it be optimized? @Hisoka-X @LJW21-02

Hisoka-X · 2025-09-28T15:05:41Z

Can it be optimized?

+1

Hisoka-X · 2025-09-28T15:11:35Z

+        <dependency>
+            <groupId>com.timecho.iotdb</groupId>
+            <artifactId>shade-iotdb-session</artifactId>
+            <version>2.0.6.1</version>
+        </dependency>


Is shade-iotdb-session not backward compatible?

LJW21-02 · 2025-09-29T09:14:03Z

@zhangshenghang Thank you again for the review suggestions! I've made all the changes mentioned above, and the code now passes all tests.

Hisoka-X

Does V1 have many users? Can we just remove it? After all, we can't support it forever.

zhangshenghang · 2025-10-06T02:16:43Z

Does V1 have many users? Can we just remove it? After all, we can't support it forever.

Please take a look at this problem。@LJW21-02

I have no other questions.

LJW21-02 · 2025-10-09T01:38:12Z

Hi @Hisoka-X ! Since V1 and V2 are not compatible and some users still rely on V1, we'd recommend keeping it for now to avoid potential issues for existing users.

CritasWang · 2025-10-09T07:32:44Z

V1 有很多用户吗？我们可以删除它吗？毕竟，我们不能永远支持它。

我们可以考虑在 Apache IoTDB 1.x 版本不再维护一段时间后移除它，但当前来说（至少 1 年内）它应该都会继续维护

davidzollo

+1

davidzollo · 2025-09-27T08:16:41Z

+
+    @Override
+    public int currentUnassignedSplitSize() {
+        return pendingSplit.size();


synchronized (stateLock) {
...
}

davidzollo · 2025-09-27T08:20:24Z

+        if (!pendingSplit.isEmpty()) {
+            assignSplit(Collections.singletonList(subtaskId));
+        }


synchronized (stateLock) {
...
}

davidzollo · 2025-09-27T08:36:50Z

+        for (int reader : readers) {
+            List<IoTDBv2SourceSplit> assignmentForReader = pendingSplit.remove(reader);
+            if (assignmentForReader != null && !assignmentForReader.isEmpty()) {
+                log.info("Assign splits {} to reader {}", assignmentForReader, reader);
+                try {
+                    context.assignSplit(reader, assignmentForReader);
+                } catch (Exception e) {
+                    log.error(
+                            "Failed to assign splits {} to reader {}",
+                            assignmentForReader,
+                            reader,
+                            e);
+                    pendingSplit.put(reader, assignmentForReader);
+                }
+            }
+        }


In this enumerator class, please guard every access to pendingSplit/shouldEnumerate with the same stateLock—wrap addPendingSplit, assignSplit, addSplitsBack, registerReader, and currentUnassignedSplitSize in that lock

LJW21-02 added 12 commits September 3, 2025 16:03

Updated session and tsfile version + Implemented sourceReader for IoT…

accb2e5

…DB-table

Extended sink connector for IoTDB-table

0c17c20

Fixed date conversion bug

c6bb8b8

Fixed connector-iotdb module spotless

fd43073

Updated both source and sink documents in ZH and EN

4a4012e

Added e2e test

b98ba95

removed comment

1e82353

fix spotless

d97621d

fixed spotless

f4ebe8b

Refactor code: seperate connectors fot IoTDB 1.X and IoTDB 2.X

b379c65

Changed shade-session + Added doc for both connectors + Modified enab…

f7b7da3

…leCompression in tableSession

fixed spotless check

4ddbaaf

github-actions Bot added document core SeaTunnel core module Zeta connectors-v2 e2e labels Sep 18, 2025

LJW21-02 and others added 3 commits September 18, 2025 10:45

Merge branch 'apache:dev' into dev

4d80b23

removed redundant log

619e2c6

Merge branch 'dev' of github.com:LJW21-02/seatunnel into dev

fec87cb

Updated from seatunnel

github-actions Bot removed the Zeta label Sep 18, 2025

zhangshenghang reviewed Sep 18, 2025

View reviewed changes

zhangshenghang reviewed Sep 19, 2025

View reviewed changes

Fixed review suggestions

aad2f96

Fixed spotless check

19cb068

davidzollo reviewed Sep 19, 2025

View reviewed changes

Comment thread ...ache/seatunnel/connectors/seatunnel/iotdbv2/sink/relational/IoTDBv2RelationalSinkClient.java

LJW21-02 added 3 commits September 23, 2025 14:21

rename

5281953

fix spotless

1cf50d7

rename fix spotless

7523afa

davidzollo reviewed Sep 24, 2025

View reviewed changes

Comment thread docs/en/connector-v2/source/IoTDBv2.md Outdated

zhangshenghang reviewed Sep 25, 2025

View reviewed changes

fix review

fc9f02f

davidzollo reviewed Sep 27, 2025

View reviewed changes

fix review suggestions

b0afaec

zhangshenghang reviewed Sep 28, 2025

View reviewed changes

Hisoka-X reviewed Sep 28, 2025

View reviewed changes

fix review suggestions

d97edff

Hisoka-X reviewed Oct 1, 2025

View reviewed changes

Hisoka-X approved these changes Oct 10, 2025

View reviewed changes

github-actions Bot added approved reviewed labels Oct 10, 2025

davidzollo approved these changes Oct 10, 2025

View reviewed changes

davidzollo merged commit 6b7db1e into apache:dev Oct 10, 2025
5 checks passed

Uh oh!

Conversation

LJW21-02 commented Sep 18, 2025

Purpose of this pull request

Does this PR introduce any user-facing change?

How was this patch tested?

Check list

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhangshenghang Sep 19, 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!

LJW21-02 commented Sep 19, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

davidzollo commented Sep 19, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

LJW21-02 commented Sep 24, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LJW21-02 commented Sep 25, 2025

Uh oh!

Hisoka-X commented Sep 26, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

davidzollo Sep 27, 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!

LJW21-02 commented Sep 28, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhangshenghang commented Sep 28, 2025

Uh oh!

Hisoka-X commented Sep 28, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

LJW21-02 commented Sep 29, 2025

Uh oh!

Hisoka-X left a comment

Choose a reason for hiding this comment

Uh oh!

zhangshenghang commented Oct 6, 2025

Uh oh!

LJW21-02 commented Oct 9, 2025

Uh oh!

CritasWang commented Oct 9, 2025

Uh oh!

davidzollo left a comment

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!

Reviewers

Assignees

Labels

Projects

zhangshenghang Sep 19, 2025 •

edited

Loading

davidzollo commented Sep 19, 2025 •

edited

Loading

davidzollo Sep 27, 2025 •

edited

Loading