This is an automated email from the ASF dual-hosted git repository.
justinchen pushed a commit to branch unknown-pipe-bug
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/unknown-pipe-bug by this push:
new a19d8d13f93 unb
a19d8d13f93 is described below
commit a19d8d13f93ee24453ab42868835659ceb8fcddd
Author: Caideyipi <[email protected]>
AuthorDate: Tue Feb 3 18:35:30 2026 +0800
unb
---
.../treemodel/auto/basic/IoTDBPipeDataSinkIT.java | 34 +++++++---------------
.../sink/protocol/legacy/IoTDBLegacyPipeSink.java | 2 +-
2 files changed, 12 insertions(+), 24 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
index 0a927cab298..035d5052cb7 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeDataSinkIT.java
@@ -23,6 +23,7 @@ import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq;
import org.apache.iotdb.db.it.utils.TestUtils;
+import org.apache.iotdb.it.env.EnvFactory;
import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic;
@@ -35,6 +36,8 @@ import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;
+import java.sql.Connection;
+import java.sql.Statement;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
@@ -211,31 +214,16 @@ public class IoTDBPipeDataSinkIT extends
AbstractPipeDualTreeModelAutoIT {
final String receiverIp = receiverDataNode.getIp();
final int receiverPort = receiverDataNode.getPort();
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute(
+ String.format(
+ "create pipe testPipe ('sink'='iotdb-legacy-pipe-sink',
'ip'='%s', 'port'='%s', 'version'='1.3')",
+ receiverIp, receiverPort));
+ }
+
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
- final Map<String, String> sourceAttributes = new HashMap<>();
- final Map<String, String> processorAttributes = new HashMap<>();
- final Map<String, String> sinkAttributes = new HashMap<>();
-
- sourceAttributes.put("source.realtime.mode", "log");
- sourceAttributes.put("user", "root");
-
- sinkAttributes.put("sink", "iotdb-legacy-pipe-sink");
- sinkAttributes.put("sink.batch.enable", "false");
- sinkAttributes.put("sink.ip", receiverIp);
- sinkAttributes.put("sink.port", Integer.toString(receiverPort));
-
- // This version does not matter since it's no longer checked by the
legacy receiver
- sinkAttributes.put("sink.version", "1.3");
-
- final TSStatus status =
- client.createPipe(
- new TCreatePipeReq("testPipe", sinkAttributes)
- .setExtractorAttributes(sourceAttributes)
- .setProcessorAttributes(processorAttributes));
-
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
-
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe("testPipe").getCode());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
index 4878facfc87..332ca6bab7c 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/legacy/IoTDBLegacyPipeSink.java
@@ -111,7 +111,7 @@ public class IoTDBLegacyPipeSink implements PipeConnector {
private String syncConnectorVersion;
private String pipeName;
- private String databaseName;
+ private String databaseName = "";
private IoTDBSyncClient client;
