Pipe: implement external sources strategy and MQTT extractor #15275

XNX02 · 2025-04-04T06:16:21Z

No description provided.

…l-source

Copilot

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

Comments suppressed due to low confidence (2)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java:185

Method name 'ExtractTable' should use lowerCamelCase to follow Java naming conventions. Consider renaming it to 'extractTable'.

private void ExtractTable(TableMessage message, MqttClientSession session) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java:276

Method name 'ExtractTree' should use lowerCamelCase to maintain consistency with Java naming conventions. Consider renaming it to 'extractTree'.

private void ExtractTree(TreeMessage message, MqttClientSession session) {

Copilot

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

SteveYurongSu · 2025-04-11T06:57:59Z

@luoluoyuyu PTAL

…urce

...mmons/src/main/java/org/apache/iotdb/commons/pipe/config/constant/PipeExtractorConstant.java

...main/java/org/apache/iotdb/confignode/procedure/impl/pipe/runtime/PipeMetaSyncProcedure.java

...src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/util/ExternalLoadBalancer.java

Caideyipi · 2025-04-15T08:07:00Z

...in/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/extractor/mqtt/MQTTExtractor.java

+
+@TreeModel
+@TableModel
+public class MQTTExtractor implements PipeExtractor {


Better add some comments here

Caideyipi · 2025-04-15T08:13:48Z

...rg/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java

@@ -2074,6 +2074,16 @@ public SettableFuture<ConfigTaskResult> alterPipe(final AlterPipeStatement alter
    final Map<String, String> connectorAttributes;
    try {
      if (!alterPipeStatement.getExtractorAttributes().isEmpty()) {
+        // We simply don't allow to alter external sources


Now the implementation is "cannot alter source if the original source is external", but it seems that the intended one is "cannot alter source type"? Now you can alter from iotdb-source to another, and cannot alter the "mqtt-source"'s own variables... Not perfect

.../datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeBuilder.java

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTExtractor.java

...-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java

...src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/util/ExternalLoadBalancer.java

...-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java

…urce

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java

Caideyipi · 2025-04-18T08:30:46Z

...re/datanode/src/main/java/org/apache/iotdb/db/pipe/resource/memory/PipeMemoryWeightUtil.java

@@ -350,6 +351,11 @@ public static long calculateAlignedChunkMetaBytesUsed(
    return size;
  }

+  public static long calculateInsertBaseStatementSizeInBytes(


Do we need to add a function here...

Caideyipi · 2025-04-18T08:47:23Z

...c/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java

+  private final PipeTabletMemoryBlock allocatedMemoryBlock;
+  private volatile ProgressIndex progressIndex;
+
+  public PipeStatementInsertionEvent(


Do we really need a "memoryBlock" for this event.. Typically this insertion event is so small that even the size of memory block that counts for the total amount of memory.

I think we can do memory control here, but using "allocatedMemoryBlock" may be to expensive, and we need to check the behavior of the "moquette" if we block the message process.

...-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java

Copilot

Pull Request Overview

This PR introduces support for external sources and adds an MQTT extractor implementation, along with related updates in task construction, load balancing, and write-back connector logic. Key changes include:

Implementation of MQTT publish handling and extraction logic.
Enhancements for external source support in pipe task building, load balancing, and meta synchronization.
Adjustments in write-back connector methods to handle event user names.

Reviewed Changes

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

Show a summary per file

File	Description
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java	Adds MQTT publish handling including table and tree message extraction.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTExtractor.java	Implements the MQTT extractor and configures the MQTT broker with custom parameters.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/tablet/PipeRawTabletInsertionEvent.java	Introduces an overloaded constructor for raw tablet insertion events.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/statement/PipeStatementInsertionEvent.java	Adds new statement insertion event functionality with reference tracking.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java	Updates write-back connector to support using an event user name and adjusts session restoration logic.
Other files	Update conditions and builder logic to support external pipe sources and load balancing.

.../src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java

XNX02 added 15 commits March 5, 2025 14:38

set up

f9225ae

Merge branch 'master' of https://github.com/apache/iotdb into externa…

6dc8c23

…l-source

update

ca62da7

Merge branch 'master' of https://github.com/apache/iotdb into externa…

67f581b

…l-source

update

ded4261

Merge branch 'master' of https://github.com/apache/iotdb into externa…

8ff357f

…l-source

update blancer

57baeb2

update

eb01779

mqtt source

3ef7238

update

5c020da

kafka source

243a376

mqtt

d61fd1a

update

ecea9a7

add header

9a82d86

update

a07e703

SteveYurongSu requested a review from Copilot April 7, 2025 02:13

SteveYurongSu self-assigned this Apr 7, 2025

Copilot AI reviewed Apr 7, 2025

View reviewed changes

fix

3dd79d6

SteveYurongSu requested a review from Copilot April 7, 2025 08:13

Copilot AI reviewed Apr 7, 2025

View reviewed changes

XNX02 added 7 commits April 12, 2025 00:09

alter pipe & loadbalancer

bf4a871

Merge branch 'master' of https://github.com/apache/iotdb into mqtt-so…

94a4ae2

…urce

update

5da03d3

update

caa3907

update

ddbdf69

update balance logic

ae41e9c

update

7b7dc32

Caideyipi reviewed Apr 15, 2025

View reviewed changes

fix

10ef051

XNX02 commented Apr 15, 2025

View reviewed changes

...src/main/java/org/apache/iotdb/confignode/procedure/impl/pipe/util/ExternalLoadBalancer.java Show resolved Hide resolved

Caideyipi reviewed Apr 15, 2025

View reviewed changes

XNX02 added 9 commits April 16, 2025 00:31

Merge branch 'master' of https://github.com/apache/iotdb into mqtt-so…

fd9932a

…urce

fix

be7aeda

update

2fa9770

update

528fd4c

fix

07f9585

update

8034b0e

update

44a55e8

fix

559ce22

fix

a331204

Caideyipi reviewed Apr 18, 2025

View reviewed changes

luoluoyuyu reviewed Apr 21, 2025

View reviewed changes

...-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTPublishHandler.java Outdated Show resolved Hide resolved

SteveYurongSu requested a review from Copilot April 21, 2025 12:06

Copilot AI reviewed Apr 21, 2025

View reviewed changes

.../src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java Outdated Show resolved Hide resolved

.../src/main/java/org/apache/iotdb/db/pipe/connector/protocol/writeback/WriteBackConnector.java Outdated Show resolved Hide resolved

XNX02 added 3 commits April 22, 2025 00:47

address review comments

556cf82

fix

9f80546

update

7b9c62d

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Pipe: implement external sources strategy and MQTT extractor #15275

Pipe: implement external sources strategy and MQTT extractor #15275

XNX02 commented Apr 4, 2025

Copilot AI left a comment

Copilot AI left a comment

SteveYurongSu commented Apr 11, 2025

Caideyipi Apr 15, 2025

Caideyipi Apr 15, 2025

Caideyipi Apr 18, 2025

Caideyipi Apr 18, 2025

Caideyipi Apr 18, 2025

Copilot AI left a comment

Pipe: implement external sources strategy and MQTT extractor #15275

Are you sure you want to change the base?

Pipe: implement external sources strategy and MQTT extractor #15275

Conversation

XNX02 commented Apr 4, 2025

Copilot AI left a comment

Choose a reason for hiding this comment

Copilot AI left a comment

Choose a reason for hiding this comment

SteveYurongSu commented Apr 11, 2025

Caideyipi Apr 15, 2025

Choose a reason for hiding this comment

Caideyipi Apr 15, 2025

Choose a reason for hiding this comment

Caideyipi Apr 18, 2025

Choose a reason for hiding this comment

Caideyipi Apr 18, 2025

Choose a reason for hiding this comment

Caideyipi Apr 18, 2025

Choose a reason for hiding this comment

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes