This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new fcaa2f82267 Pipe: Reuse clusters across related integration tests
(#18400)
fcaa2f82267 is described below
commit fcaa2f82267189eae4300351f4c8ac9945ce615b
Author: Caideyipi <[email protected]>
AuthorDate: Fri Aug 7 15:17:04 2026 +0800
Pipe: Reuse clusters across related integration tests (#18400)
---
.../auto/basic/IoTDBPipeTwoStageAggregateIT.java | 146 -----------
.../manual/IoTDBPipeManualConflictIT.java | 70 +----
.../treemodel/manual/IoTDBPipePermissionIT.java | 147 +++++++++--
.../IoTDBLegacyPipeReceiverAutoCreateIT.java | 159 ------------
...verSecurityIT.java => IoTDBPipeReceiverIT.java} | 249 +++++++++++++++++-
.../pipe/it/single/IoTDBPipeReceiverSessionIT.java | 281 ---------------------
6 files changed, 371 insertions(+), 681 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeTwoStageAggregateIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeTwoStageAggregateIT.java
deleted file mode 100644
index ac588aa91e8..00000000000
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeTwoStageAggregateIT.java
+++ /dev/null
@@ -1,146 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.pipe.it.dual.treemodel.auto.basic;
-
-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.MultiEnvFactory;
-import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
-import org.apache.iotdb.it.framework.IoTDBTestRunner;
-import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeAutoBasic;
-import
org.apache.iotdb.pipe.it.dual.treemodel.auto.AbstractPipeDualTreeModelAutoIT;
-import org.apache.iotdb.rpc.TSStatusCode;
-
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-
-@RunWith(IoTDBTestRunner.class)
-@Category({MultiClusterIT2DualTreeAutoBasic.class})
-public class IoTDBPipeTwoStageAggregateIT extends
AbstractPipeDualTreeModelAutoIT {
-
- @Override
- @Before
- public void setUp() {
- MultiEnvFactory.createEnv(2);
- senderEnv = MultiEnvFactory.getEnv(0);
- receiverEnv = MultiEnvFactory.getEnv(1);
- setupConfig();
- senderEnv.initClusterEnvironment(1, 1);
- receiverEnv.initClusterEnvironment(1, 1);
- }
-
- @Override
- protected void setupConfig() {
- super.setupConfig();
- senderEnv
- .getConfig()
- .getCommonConfig()
- .setDataReplicationFactor(1)
- .setSchemaReplicationFactor(1);
- receiverEnv
- .getConfig()
- .getCommonConfig()
- .setAutoCreateSchemaEnabled(false)
- .setDataReplicationFactor(1)
- .setSchemaReplicationFactor(1);
- }
-
- @Test
- public void testCountPointProcessorUsesSourceCredentials() throws Exception {
- final String sourceUser = "countPointUser";
- final String sourcePassword = "StrngPsWd@623454";
- final String sourceDevice = "root.twostage_source.d1";
- final String processorOutputSeries =
"root.twostage_source.result.point_count";
- // The processor uses the configured output series as the tablet device
and its measurement
- // node as the tablet measurement.
- final String outputDevice = processorOutputSeries;
- final String outputSeries = outputDevice + ".point_count";
-
- TestUtils.executeNonQueries(
- senderEnv,
- Arrays.asList(
- "CREATE DATABASE root.twostage_source",
- "CREATE TIMESERIES " + sourceDevice + ".s1 WITH
DATATYPE=INT32,ENCODING=RLE",
- "CREATE USER " + sourceUser + " '" + sourcePassword + "'",
- "GRANT READ_DATA ON " + sourceDevice + ".s1 TO USER " +
sourceUser),
- null);
- TestUtils.executeNonQueries(
- receiverEnv,
- Arrays.asList(
- "CREATE DATABASE root.twostage_source",
- "CREATE TIMESERIES " + outputSeries + " WITH
DATATYPE=INT64,ENCODING=RLE",
- "INSERT INTO " + outputDevice + "(time,point_count) VALUES (0,0)"),
- null);
-
- final Map<String, String> sourceAttributes = new HashMap<>();
- sourceAttributes.put("source.path", sourceDevice + ".s1");
- sourceAttributes.put("source.watermark.interval-ms", "500");
- sourceAttributes.put("user", sourceUser);
- sourceAttributes.put("password", sourcePassword);
-
- final Map<String, String> processorAttributes = new HashMap<>();
- processorAttributes.put("processor", "count-point-processor");
- processorAttributes.put("processor.output.series", processorOutputSeries);
-
- final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
- final Map<String, String> sinkAttributes = new HashMap<>();
- sinkAttributes.put("sink", "iotdb-thrift-sink");
- sinkAttributes.put("sink.batch.enable", "false");
- sinkAttributes.put("sink.ip", receiverDataNode.getIp());
- sinkAttributes.put("sink.port",
Integer.toString(receiverDataNode.getPort()));
- sinkAttributes.put("sink.user", "root");
- sinkAttributes.put("sink.password", "root");
-
- try (final SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
- final TSStatus createStatus =
- client.createPipe(
- new TCreatePipeReq("countPointPipe", sinkAttributes)
- .setExtractorAttributes(sourceAttributes)
- .setProcessorAttributes(processorAttributes));
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
createStatus.getCode());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- client.startPipe("countPointPipe").getCode());
- }
-
- TestUtils.executeNonQueries(
- senderEnv,
- Collections.singletonList(
- "INSERT INTO " + sourceDevice + "(time,s1) VALUES
(1,1),(2,2),(3,3)"),
- null);
-
- TestUtils.assertDataEventuallyOnEnv(
- receiverEnv,
- "SELECT MAX_VALUE(point_count) FROM " + outputDevice,
- "MAX_VALUE(" + outputSeries + "),",
- Collections.singleton("3,"));
- }
-}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeManualConflictIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeManualConflictIT.java
index 76b00816606..e168375ca00 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeManualConflictIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipeManualConflictIT.java
@@ -43,7 +43,7 @@ import java.util.Map;
@Category({MultiClusterIT2DualTreeManual.class})
public class IoTDBPipeManualConflictIT extends
AbstractPipeDualTreeModelManualIT {
@Test
- public void testDoubleLivingTimeseries() throws Exception {
+ public void testDoubleLivingTimeseriesAndTemplate() throws Exception {
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
final String receiverIp = receiverDataNode.getIp();
@@ -149,73 +149,11 @@ public class IoTDBPipeManualConflictIT extends
AbstractPipeDualTreeModelManualIT
"select count(*) from root.ln.** group by level=1",
"count(root.ln.*.*.*),",
Collections.singleton("2,"));
- }
-
- @Test
- public void testDoubleLivingTemplate() throws Exception {
- final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
-
- final String receiverIp = receiverDataNode.getIp();
- final int receiverPort = receiverDataNode.getPort();
-
- 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.inclusion", "data, schema");
- sourceAttributes.put("source.forwarding-pipe-requests", "false");
- sourceAttributes.put("user", "root");
- sinkAttributes.put("sink", "iotdb-thrift-sink");
- sinkAttributes.put("sink.exception.conflict.resolve-strategy", "retry");
- sinkAttributes.put("sink.exception.conflict.retry-max-time-seconds",
"-1");
- sinkAttributes.put("sink.batch.enable", "false");
- sinkAttributes.put("sink.ip", receiverIp);
- sinkAttributes.put("sink.port", Integer.toString(receiverPort));
-
- 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());
- }
-
- try (final SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
receiverEnv.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.inclusion", "data, schema");
- sourceAttributes.put("source.forwarding-pipe-requests", "false");
- sourceAttributes.put("user", "root");
-
- sinkAttributes.put("sink", "iotdb-thrift-sink");
- sinkAttributes.put("sink.exception.conflict.resolve-strategy", "retry");
- sinkAttributes.put("sink.exception.conflict.retry-max-time-seconds",
"-1");
- sinkAttributes.put("sink.batch.enable", "false");
- sinkAttributes.put("sink.ip", senderEnv.getDataNodeWrapper(0).getIp());
- sinkAttributes.put("sink.port",
Integer.toString(senderEnv.getDataNodeWrapper(0).getPort()));
-
- 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());
- }
+ assertDoubleLivingTemplate();
+ }
+ private void assertDoubleLivingTemplate() throws Exception {
TestUtils.executeNonQueries(
senderEnv,
Arrays.asList(
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java
index 4548c12cef3..f2d5faa6e93 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/manual/IoTDBPipePermissionIT.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.MultiClusterIT2DualTreeManual;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.awaitility.Awaitility;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
@@ -46,6 +47,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import static org.junit.Assert.fail;
@@ -94,16 +96,7 @@ public class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
}
@Test
- public void testWithSyncSink() throws Exception {
- testWithSink("iotdb-thrift-sync-sink");
- }
-
- @Test
- public void testWithAsyncSink() throws Exception {
- testWithSink("iotdb-thrift-async-sink");
- }
-
- private void testWithSink(final String sink) throws Exception {
+ public void testWithSyncAndAsyncSink() throws Exception {
TestUtils.executeNonQueries(
receiverEnv,
Arrays.asList(
@@ -113,6 +106,15 @@ public class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
"grant WRITE, READ, SYSTEM, SECURITY on root.** to role `admin`"),
null);
+ testWithSink("iotdb-thrift-sync-sink", "sync");
+ testWithSink("iotdb-thrift-async-sink", "async");
+ }
+
+ private void testWithSink(final String sink, final String suffix) throws
Exception {
+ final String sourceUser = "user_" + suffix;
+ final String pipeName = "testPipe_" + suffix;
+ final String device = "root.ln." + suffix + ".wt01";
+
final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
final String receiverIp = receiverDataNode.getIp();
final int receiverPort = receiverDataNode.getPort();
@@ -122,10 +124,12 @@ public class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
TestUtils.executeNonQueries(
senderEnv,
Arrays.asList(
- "create user user 'passwd123456'",
- "create timeseries root.ln.wf02.wt01.temperature with
datatype=INT64,encoding=PLAIN",
- "create timeseries root.ln.wf02.wt01.status with
datatype=BOOLEAN,encoding=PLAIN",
- "insert into root.ln.wf02.wt01(time, temperature, status) values
(1800000000000, 23, true)"),
+ "create user " + sourceUser + " 'passwd123456'",
+ "create timeseries " + device + ".temperature with
datatype=INT64,encoding=PLAIN",
+ "create timeseries " + device + ".status with
datatype=BOOLEAN,encoding=PLAIN",
+ "insert into "
+ + device
+ + "(time, temperature, status) values (1800000000000, 23,
true)"),
null);
awaitUntilFlush(senderEnv);
@@ -144,39 +148,59 @@ public class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
final TSStatus status =
client.createPipe(
- new TCreatePipeReq("testPipe", sinkAttributes)
+ new TCreatePipeReq(pipeName, sinkAttributes)
.setExtractorAttributes(sourceAttributes)
.setProcessorAttributes(processorAttributes));
Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
status.getCode());
Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe("testPipe").getCode());
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.startPipe(pipeName).getCode());
- TestUtils.assertDataEventuallyOnEnv(
- receiverEnv,
- "list user",
- "UserId,User,",
- new HashSet<>(Arrays.asList("0,root,", "10001,user,",
"10000,thulab,")));
+ assertUserEventuallyExists(sourceUser);
final Set<String> expectedResSet = new HashSet<>();
expectedResSet.add(
-
"root.ln.wf02.wt01.temperature,null,root.ln,INT64,PLAIN,LZ4,null,null,null,null,BASE,");
+ device +
".temperature,null,root.ln,INT64,PLAIN,LZ4,null,null,null,null,BASE,");
expectedResSet.add(
-
"root.ln.wf02.wt01.status,null,root.ln,BOOLEAN,PLAIN,LZ4,null,null,null,null,BASE,");
+ device +
".status,null,root.ln,BOOLEAN,PLAIN,LZ4,null,null,null,null,BASE,");
TestUtils.assertDataEventuallyOnEnv(
receiverEnv,
- "show timeseries root.ln.**",
+ "show timeseries " + device + ".**",
"Timeseries,Alias,Database,DataType,Encoding,Compression,Tags,Attributes,Deadband,DeadbandParameters,ViewType,",
expectedResSet);
expectedResSet.clear();
TestUtils.assertDataEventuallyOnEnv(
receiverEnv,
- "select * from root.ln.**",
- "Time,root.ln.wf02.wt01.temperature,root.ln.wf02.wt01.status,",
+ "select * from " + device,
+ "Time," + device + ".temperature," + device + ".status,",
Collections.singleton("1800000000000,23,true,"));
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
client.dropPipe(pipeName).getCode());
}
}
+ private void assertUserEventuallyExists(final String username) {
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollInterval(1, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ try (final Connection connection = receiverEnv.getConnection();
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet = statement.executeQuery("list
user")) {
+ boolean found = false;
+ while (resultSet.next()) {
+ if (username.equals(resultSet.getString("User"))) {
+ found = true;
+ break;
+ }
+ }
+ Assert.assertTrue(found);
+ }
+ });
+ }
+
@Test
public void testNoPermission() throws Exception {
TestUtils.executeNonQueries(
@@ -238,7 +262,7 @@ public class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
}
@Test
- public void testSourcePermissionRestart() throws SQLException {
+ public void testSourcePermissionForDataAndTwoStageProcessor() throws
Exception {
try (final Connection connection = senderEnv.getConnection();
final Statement statement = connection.createStatement()) {
TestUtils.executeNonQuery(senderEnv, "create user `thulab`
'passwD@123456'", connection);
@@ -277,6 +301,75 @@ public class IoTDBPipePermissionIT extends
AbstractPipeDualTreeModelManualIT {
"count(root.vehicle.plane.pressure),",
Collections.singleton("1,"));
}
+
+ testCountPointProcessorUsesSourceCredentials();
+ }
+
+ private void testCountPointProcessorUsesSourceCredentials() throws Exception
{
+ final String sourceDevice = "root.twostage_source.d1";
+ final String processorOutputSeries =
"root.twostage_source.result.point_count";
+ // The processor uses the configured output series as the tablet device
and its measurement
+ // node as the tablet measurement.
+ final String outputDevice = processorOutputSeries;
+ final String outputSeries = outputDevice + ".point_count";
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Arrays.asList(
+ "CREATE DATABASE root.twostage_source",
+ "CREATE TIMESERIES " + sourceDevice + ".s1 WITH
DATATYPE=INT32,ENCODING=RLE"),
+ null);
+ TestUtils.executeNonQueries(
+ receiverEnv,
+ Arrays.asList(
+ "CREATE DATABASE root.twostage_source",
+ "CREATE TIMESERIES " + outputSeries + " WITH
DATATYPE=INT64,ENCODING=RLE",
+ "INSERT INTO " + outputDevice + "(time,point_count) VALUES (0,0)"),
+ null);
+
+ final Map<String, String> sourceAttributes = new HashMap<>();
+ sourceAttributes.put("source.path", sourceDevice + ".s1");
+ sourceAttributes.put("source.watermark.interval-ms", "500");
+ sourceAttributes.put("user", "thulab");
+ sourceAttributes.put("password", "passwD@123456");
+
+ final Map<String, String> processorAttributes = new HashMap<>();
+ processorAttributes.put("processor", "count-point-processor");
+ processorAttributes.put("processor.output.series", processorOutputSeries);
+
+ final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0);
+ final Map<String, String> sinkAttributes = new HashMap<>();
+ sinkAttributes.put("sink", "iotdb-thrift-sink");
+ sinkAttributes.put("sink.batch.enable", "false");
+ sinkAttributes.put("sink.ip", receiverDataNode.getIp());
+ sinkAttributes.put("sink.port",
Integer.toString(receiverDataNode.getPort()));
+ sinkAttributes.put("sink.user", "root");
+ sinkAttributes.put("sink.password", "root");
+
+ try (final SyncConfigNodeIServiceClient client =
+ (SyncConfigNodeIServiceClient)
senderEnv.getLeaderConfigNodeConnection()) {
+ final TSStatus createStatus =
+ client.createPipe(
+ new TCreatePipeReq("countPointPipe", sinkAttributes)
+ .setExtractorAttributes(sourceAttributes)
+ .setProcessorAttributes(processorAttributes));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
createStatus.getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client.startPipe("countPointPipe").getCode());
+ }
+
+ TestUtils.executeNonQueries(
+ senderEnv,
+ Collections.singletonList(
+ "INSERT INTO " + sourceDevice + "(time,s1) VALUES
(1,1),(2,2),(3,3)"),
+ null);
+
+ TestUtils.assertDataEventuallyOnEnv(
+ receiverEnv,
+ "SELECT MAX_VALUE(point_count) FROM " + outputDevice,
+ "MAX_VALUE(" + outputSeries + "),",
+ Collections.singleton("3,"));
}
@Test
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverAutoCreateIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverAutoCreateIT.java
deleted file mode 100644
index 3325c963b22..00000000000
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverAutoCreateIT.java
+++ /dev/null
@@ -1,159 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.pipe.it.single;
-
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.property.ThriftClientProperty;
-import org.apache.iotdb.commons.conf.IoTDBConstant;
-import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient;
-import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData;
-import org.apache.iotdb.isession.SessionConfig;
-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.it.utils.TsFileGenerator;
-import org.apache.iotdb.itbase.category.LocalStandaloneIT;
-import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
-import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
-import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
-import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
-import org.apache.iotdb.service.rpc.thrift.TSyncIdentityInfo;
-import org.apache.iotdb.service.rpc.thrift.TSyncTransportMetaInfo;
-
-import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.external.commons.io.FileUtils;
-import org.apache.tsfile.file.metadata.enums.TSEncoding;
-import org.apache.tsfile.write.schema.MeasurementSchema;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-
-import java.io.File;
-import java.nio.ByteBuffer;
-import java.nio.file.Files;
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.Statement;
-import java.time.ZoneId;
-import java.util.Collections;
-
-@RunWith(IoTDBTestRunner.class)
-@Category({LocalStandaloneIT.class})
-public class IoTDBLegacyPipeReceiverAutoCreateIT {
-
- private static final String DATABASE = "root.legacy_no_auto_create";
- private static final String DEVICE = DATABASE + ".d1";
- private static final String TSFILE_NAME = "0-" + DATABASE +
"-0-0-0-0-0-0.tsfile";
-
- @BeforeClass
- public static void setUp() {
- EnvFactory.getEnv()
- .getConfig()
- .getCommonConfig()
- .setAutoCreateSchemaEnabled(false)
- .setDatanodeMemoryProportion("3:3:1:1:1:0");
- EnvFactory.getEnv().initClusterEnvironment();
- }
-
- @AfterClass
- public static void tearDown() {
- EnvFactory.getEnv().cleanClusterEnvironment();
- }
-
- @Test
- public void testLegacyHandshakeAndTsFileLoadRespectDisabledAutoCreate()
throws Exception {
- final File tempDir =
Files.createTempDirectory("legacy-pipe-no-auto-create").toFile();
- try {
- final File tsFile = new File(tempDir, TSFILE_NAME);
- try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) {
- generator.registerTimeseries(
- DEVICE,
- Collections.singletonList(
- new MeasurementSchema("s1", TSDataType.INT32,
TSEncoding.RLE)));
- generator.generateData(DEVICE, 2, 1, false);
- }
-
- final DataNodeWrapper dataNode =
EnvFactory.getEnv().getDataNodeWrapper(0);
- try (final IoTDBSyncClient client =
- new IoTDBSyncClient(
- new ThriftClientProperty.Builder().build(),
- dataNode.getIp(),
- dataNode.getPort(),
- false,
- null,
- null)) {
- final TSOpenSessionResp openSessionResp =
client.openSession(createOpenSessionReq());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
openSessionResp.getStatus().getCode());
-
- try {
- final TSStatus handshakeStatus =
- client.handshake(
- new TSyncIdentityInfo(
- "legacyNoAutoCreate", System.currentTimeMillis(),
"UNKNOWN", DATABASE));
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
handshakeStatus.getCode());
- assertDatabaseDoesNotExist();
-
- final TSStatus fileStatus =
- client.sendFile(
- new TSyncTransportMetaInfo(tsFile.getName(), 0),
- ByteBuffer.wrap(Files.readAllBytes(tsFile.toPath())));
- Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
fileStatus.getCode());
-
- final TSStatus loadStatus =
- client.sendPipeData(
- ByteBuffer.wrap(new TsFilePipeData("", tsFile.getName(),
1).serialize()));
- Assert.assertEquals(TSStatusCode.PIPESERVER_ERROR.getStatusCode(),
loadStatus.getCode());
- assertDatabaseDoesNotExist();
- } finally {
- client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
- }
- }
- } finally {
- FileUtils.deleteDirectory(tempDir);
- }
- }
-
- private void assertDatabaseDoesNotExist() throws Exception {
- try (final Connection connection = EnvFactory.getEnv().getConnection();
- final Statement statement = connection.createStatement();
- final ResultSet resultSet = statement.executeQuery("SHOW DATABASES")) {
- while (resultSet.next()) {
- Assert.assertNotEquals(DATABASE, resultSet.getString(1));
- }
- }
- }
-
- private TSOpenSessionReq createOpenSessionReq() {
- final TSOpenSessionReq req = new TSOpenSessionReq();
- req.setClient_protocol(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3);
- req.setUsername(SessionConfig.DEFAULT_USER);
- req.setPassword(SessionConfig.DEFAULT_PASSWORD);
- req.setZoneId(ZoneId.systemDefault().toString());
- req.putToConfiguration("version",
IoTDBConstant.ClientVersion.V_1_0.toString());
- req.putToConfiguration("sql_dialect", "tree");
- return req;
- }
-}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverIT.java
similarity index 58%
rename from
integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
rename to
integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverIT.java
index d383ab3de8a..5ca62ebb15f 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverSecurityIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverIT.java
@@ -21,9 +21,18 @@ package org.apache.iotdb.pipe.it.single;
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.property.ThriftClientProperty;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.path.MeasurementPath;
import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient;
+import
org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferHandshakeConstant;
+import
org.apache.iotdb.commons.pipe.sink.payload.thrift.request.IoTDBSinkRequestVersion;
+import
org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
+import
org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigNodeHandshakeV1Req;
+import
org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigNodeHandshakeV2Req;
+import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
+import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
+import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReq;
import org.apache.iotdb.db.pipe.sink.payload.legacy.PipeData;
import org.apache.iotdb.db.pipe.sink.payload.legacy.TsFilePipeData;
import org.apache.iotdb.db.storageengine.dataregion.modification.v1.Deletion;
@@ -34,6 +43,9 @@ import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.it.utils.TsFileGenerator;
import org.apache.iotdb.itbase.category.LocalStandaloneIT;
import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeReq;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
+import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
@@ -45,6 +57,7 @@ import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.external.commons.io.FileUtils;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
import org.apache.tsfile.utils.ReadWriteIOUtils;
+import org.apache.tsfile.write.record.Tablet;
import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.AfterClass;
import org.junit.Assert;
@@ -65,10 +78,15 @@ import java.sql.SQLException;
import java.sql.Statement;
import java.time.ZoneId;
import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class})
-public class IoTDBLegacyPipeReceiverSecurityIT {
+public class IoTDBPipeReceiverIT {
+
+ private static final String DATA_NODE_DEVICE =
"root.pipe_receiver_session.d1";
private static final String LEGACY_PIPE_USER = "pipeHack";
private static final String LEGACY_PIPE_PASSWORD = "StrngPsWd@623451";
@@ -85,9 +103,20 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
private static final String LEGACY_TSFILE_NAME =
"0-" + LEGACY_TSFILE_DATABASE + "-0-0-0-0-0-0.tsfile";
+ private static final String NO_AUTO_CREATE_DATABASE =
"root.legacy_no_auto_create";
+ private static final String NO_AUTO_CREATE_DEVICE = NO_AUTO_CREATE_DATABASE
+ ".d1";
+ private static final String NO_AUTO_CREATE_TSFILE_NAME =
+ "0-" + NO_AUTO_CREATE_DATABASE + "-0-0-0-0-0-0.tsfile";
+
@BeforeClass
public static void setUp() {
-
EnvFactory.getEnv().getConfig().getCommonConfig().setDatanodeMemoryProportion("3:3:1:1:1:0");
+ EnvFactory.getEnv()
+ .getConfig()
+ .getCommonConfig()
+ .setAutoCreateSchemaEnabled(false)
+ .setDatanodeMemoryProportion("3:3:1:1:1:0")
+ .setPipeMemoryManagementEnabled(false)
+ .setIsPipeEnableMemoryCheck(false);
EnvFactory.getEnv().initClusterEnvironment();
}
@@ -96,6 +125,140 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
EnvFactory.getEnv().cleanClusterEnvironment();
}
+ @Test
+ public void testDataNodeReceiverSessionHandling() throws Exception {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE root.pipe_receiver_session");
+ statement.execute(
+ "CREATE TIMESERIES " + DATA_NODE_DEVICE + ".s1 WITH
DATATYPE=INT32,ENCODING=RLE");
+ }
+
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ try (final IoTDBSyncClient client = createClient(dataNode)) {
+ Assert.assertEquals(
+ TSStatusCode.NOT_LOGIN.getStatusCode(),
+ client.pipeTransfer(buildTabletReq(1, 1)).getStatus().getCode());
+ Assert.assertEquals(
+ TSStatusCode.NOT_LOGIN.getStatusCode(),
+ client.getBackupConfiguration().getStatus().getCode());
+
Assert.assertTrue(client.fetchAllConnectionsInfo().getConnectionInfoList().isEmpty());
+ Assert.assertEquals(
+ TSStatusCode.NOT_LOGIN.getStatusCode(),
+ client
+ .pipeSubscribe(new TPipeSubscribeReq().setVersion((byte)
1).setType((short) 0))
+ .getStatus()
+ .getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.PIPE_HANDSHAKE_ERROR.getStatusCode(),
+ client
+ .pipeTransfer(
+ PipeTransferDataNodeHandshakeV1Req.toTPipeTransferReq(
+
CommonDescriptor.getInstance().getConfig().getTimestampPrecision()))
+ .getStatus()
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.NOT_LOGIN.getStatusCode(),
+ client
+ .pipeTransfer(
+ PipeTransferDataNodeHandshakeV2Req.toTPipeTransferReq(
+ buildHandshakeParams(null, null)))
+ .getStatus()
+ .getCode());
+
+ final TPipeTransferResp wrongPasswordResp =
+ client.pipeTransfer(
+ PipeTransferDataNodeHandshakeV2Req.toTPipeTransferReq(
+ buildHandshakeParams(SessionConfig.DEFAULT_USER,
"wrong-password")));
+ Assert.assertNotEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
wrongPasswordResp.getStatus().getCode());
+ Assert.assertEquals(
+ TSStatusCode.NOT_LOGIN.getStatusCode(),
+ client.pipeTransfer(buildTabletReq(1, 1)).getStatus().getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client
+ .pipeTransfer(
+ PipeTransferDataNodeHandshakeV2Req.toTPipeTransferReq(
+ buildHandshakeParams(
+ SessionConfig.DEFAULT_USER,
SessionConfig.DEFAULT_PASSWORD)))
+ .getStatus()
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client.pipeTransfer(buildTabletReq(1, 1)).getStatus().getCode());
+ }
+
+ try (final IoTDBSyncClient client = createClient(dataNode)) {
+ final TSOpenSessionResp openSessionResp =
client.openSession(createOpenSessionReq());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
openSessionResp.getStatus().getCode());
+ try {
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client.getBackupConfiguration().getStatus().getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client.pipeTransfer(buildTabletReq(2, 2)).getStatus().getCode());
+ } finally {
+ client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
+ }
+ }
+
+ assertTimeseriesRowCount(DATA_NODE_DEVICE, "s1", 2);
+ }
+
+ @Test
+ public void testConfigNodeReceiverSessionHandling() throws Exception {
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ try (final IoTDBSyncClient client = createClient(dataNode)) {
+ Assert.assertEquals(
+ TSStatusCode.PIPE_CONFIG_RECEIVER_HANDSHAKE_NEEDED.getStatusCode(),
+
client.pipeTransfer(buildEmptyConfigPlanReq()).getStatus().getCode());
+ Assert.assertEquals(
+ TSStatusCode.PIPE_HANDSHAKE_ERROR.getStatusCode(),
+ client
+ .pipeTransfer(
+ PipeTransferConfigNodeHandshakeV1Req.toTPipeTransferReq(
+
CommonDescriptor.getInstance().getConfig().getTimestampPrecision()))
+ .getStatus()
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.NOT_LOGIN.getStatusCode(),
+ client
+ .pipeTransfer(
+ PipeTransferConfigNodeHandshakeV2Req.toTPipeTransferReq(
+ buildHandshakeParams(null, null)))
+ .getStatus()
+ .getCode());
+
+ final TPipeTransferResp wrongPasswordResp =
+ client.pipeTransfer(
+ PipeTransferConfigNodeHandshakeV2Req.toTPipeTransferReq(
+ buildHandshakeParams(SessionConfig.DEFAULT_USER,
"wrong-password")));
+ Assert.assertNotEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
wrongPasswordResp.getStatus().getCode());
+ Assert.assertEquals(
+ TSStatusCode.PIPE_CONFIG_RECEIVER_HANDSHAKE_NEEDED.getStatusCode(),
+
client.pipeTransfer(buildEmptyConfigPlanReq()).getStatus().getCode());
+
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ client
+ .pipeTransfer(
+ PipeTransferConfigNodeHandshakeV2Req.toTPipeTransferReq(
+ buildHandshakeParams(
+ SessionConfig.DEFAULT_USER,
SessionConfig.DEFAULT_PASSWORD)))
+ .getStatus()
+ .getCode());
+ Assert.assertNotEquals(
+ TSStatusCode.NOT_LOGIN.getStatusCode(),
+
client.pipeTransfer(buildEmptyConfigPlanReq()).getStatus().getCode());
+ }
+ }
+
@Test
public void testRejectPathTraversalFileNameInLegacyTransportFile() throws
Exception {
final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
@@ -242,6 +405,78 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
assertTimeseriesRowCount(LEGACY_TSFILE_DEVICE, "s1", 0);
}
+ @Test
+ public void testLegacyHandshakeAndTsFileLoadRespectDisabledAutoCreate()
throws Exception {
+ final File tempDir =
Files.createTempDirectory("legacy-pipe-no-auto-create").toFile();
+ try {
+ final File tsFile = new File(tempDir, NO_AUTO_CREATE_TSFILE_NAME);
+ generateTsFile(tsFile, NO_AUTO_CREATE_DEVICE);
+
+ final DataNodeWrapper dataNode =
EnvFactory.getEnv().getDataNodeWrapper(0);
+ try (final IoTDBSyncClient client = createClient(dataNode)) {
+ final TSOpenSessionResp openSessionResp =
client.openSession(createOpenSessionReq());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
openSessionResp.getStatus().getCode());
+
+ try {
+ final TSStatus handshakeStatus =
+ client.handshake(
+ new TSyncIdentityInfo(
+ "legacyNoAutoCreate",
+ System.currentTimeMillis(),
+ "UNKNOWN",
+ NO_AUTO_CREATE_DATABASE));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
handshakeStatus.getCode());
+ assertDatabaseDoesNotExist(NO_AUTO_CREATE_DATABASE);
+
+ final TSStatus loadStatus = sendLegacyTsFile(client, tsFile);
+ Assert.assertEquals(TSStatusCode.PIPESERVER_ERROR.getStatusCode(),
loadStatus.getCode());
+ assertDatabaseDoesNotExist(NO_AUTO_CREATE_DATABASE);
+ } finally {
+ client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
+ }
+ }
+ } finally {
+ FileUtils.deleteDirectory(tempDir);
+ }
+ }
+
+ private TPipeTransferReq buildEmptyConfigPlanReq() {
+ return new TPipeTransferReq()
+ .setVersion(IoTDBSinkRequestVersion.VERSION_1.getVersion())
+ .setType(PipeRequestType.TRANSFER_CONFIG_PLAN.getType())
+ .setBody(ByteBuffer.allocate(0));
+ }
+
+ private TPipeTransferReq buildTabletReq(final long timestamp, final int
value) throws Exception {
+ final Tablet tablet =
+ new Tablet(
+ DATA_NODE_DEVICE,
+ Collections.singletonList(new MeasurementSchema("s1",
TSDataType.INT32)),
+ 1);
+ tablet.addTimestamp(0, timestamp);
+ tablet.addValue("s1", 0, value);
+ return PipeTransferTabletRawReq.toTPipeTransferReq(tablet, false);
+ }
+
+ private Map<String, String> buildHandshakeParams(final String username,
final String password) {
+ final Map<String, String> params = new HashMap<>();
+ params.put(
+ PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLUSTER_ID,
+ "pipe-session-it-" + UUID.randomUUID());
+ params.put(
+ PipeTransferHandshakeConstant.HANDSHAKE_KEY_TIME_PRECISION,
+ CommonDescriptor.getInstance().getConfig().getTimestampPrecision());
+ if (username != null) {
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USERNAME,
username);
+ }
+ if (password != null) {
+ params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_PASSWORD,
password);
+ }
+ return params;
+ }
+
private void assertLegacyPipeRpcStatus(
final IoTDBSyncClient client, final TSStatusCode expectedStatusCode)
throws Exception {
final int expectedCode = expectedStatusCode.getStatusCode();
@@ -320,6 +555,16 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
}
}
+ private void assertDatabaseDoesNotExist(final String database) throws
SQLException {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet = statement.executeQuery("SHOW DATABASES")) {
+ while (resultSet.next()) {
+ Assert.assertNotEquals(database, resultSet.getString(1));
+ }
+ }
+ }
+
private void generateTsFile(final File tsFile, final String device) throws
Exception {
try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) {
generator.registerTimeseries(
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverSessionIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverSessionIT.java
deleted file mode 100644
index a1f98723e6b..00000000000
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverSessionIT.java
+++ /dev/null
@@ -1,281 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.iotdb.pipe.it.single;
-
-import org.apache.iotdb.commons.client.property.ThriftClientProperty;
-import org.apache.iotdb.commons.conf.CommonDescriptor;
-import org.apache.iotdb.commons.conf.IoTDBConstant;
-import org.apache.iotdb.commons.pipe.sink.client.IoTDBSyncClient;
-import
org.apache.iotdb.commons.pipe.sink.payload.thrift.common.PipeTransferHandshakeConstant;
-import
org.apache.iotdb.commons.pipe.sink.payload.thrift.request.IoTDBSinkRequestVersion;
-import
org.apache.iotdb.commons.pipe.sink.payload.thrift.request.PipeRequestType;
-import
org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigNodeHandshakeV1Req;
-import
org.apache.iotdb.confignode.manager.pipe.sink.payload.PipeTransferConfigNodeHandshakeV2Req;
-import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV1Req;
-import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferDataNodeHandshakeV2Req;
-import
org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTabletRawReq;
-import org.apache.iotdb.isession.SessionConfig;
-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.LocalStandaloneIT;
-import org.apache.iotdb.rpc.TSStatusCode;
-import org.apache.iotdb.service.rpc.thrift.TPipeSubscribeReq;
-import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
-import org.apache.iotdb.service.rpc.thrift.TPipeTransferResp;
-import org.apache.iotdb.service.rpc.thrift.TSCloseSessionReq;
-import org.apache.iotdb.service.rpc.thrift.TSOpenSessionReq;
-import org.apache.iotdb.service.rpc.thrift.TSOpenSessionResp;
-import org.apache.iotdb.service.rpc.thrift.TSProtocolVersion;
-
-import org.apache.tsfile.enums.TSDataType;
-import org.apache.tsfile.write.record.Tablet;
-import org.apache.tsfile.write.schema.MeasurementSchema;
-import org.junit.AfterClass;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-
-import java.nio.ByteBuffer;
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.Statement;
-import java.time.ZoneId;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
-@RunWith(IoTDBTestRunner.class)
-@Category({LocalStandaloneIT.class})
-public class IoTDBPipeReceiverSessionIT {
-
- private static final String DATA_NODE_DEVICE =
"root.pipe_receiver_session.d1";
-
- @BeforeClass
- public static void setUp() {
- EnvFactory.getEnv()
- .getConfig()
- .getCommonConfig()
- .setDatanodeMemoryProportion("3:3:1:1:1:0")
- .setPipeMemoryManagementEnabled(false)
- .setIsPipeEnableMemoryCheck(false);
- EnvFactory.getEnv().initClusterEnvironment();
- }
-
- @AfterClass
- public static void tearDown() {
- EnvFactory.getEnv().cleanClusterEnvironment();
- }
-
- @Test
- public void testDataNodeReceiverSessionHandling() throws Exception {
- final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
- try (final IoTDBSyncClient client = createClient(dataNode)) {
- Assert.assertEquals(
- TSStatusCode.NOT_LOGIN.getStatusCode(),
- client.pipeTransfer(buildTabletReq(1, 1)).getStatus().getCode());
- Assert.assertEquals(
- TSStatusCode.NOT_LOGIN.getStatusCode(),
- client.getBackupConfiguration().getStatus().getCode());
-
Assert.assertTrue(client.fetchAllConnectionsInfo().getConnectionInfoList().isEmpty());
- Assert.assertEquals(
- TSStatusCode.NOT_LOGIN.getStatusCode(),
- client
- .pipeSubscribe(new TPipeSubscribeReq().setVersion((byte)
1).setType((short) 0))
- .getStatus()
- .getCode());
-
- Assert.assertEquals(
- TSStatusCode.PIPE_HANDSHAKE_ERROR.getStatusCode(),
- client
- .pipeTransfer(
- PipeTransferDataNodeHandshakeV1Req.toTPipeTransferReq(
-
CommonDescriptor.getInstance().getConfig().getTimestampPrecision()))
- .getStatus()
- .getCode());
- Assert.assertEquals(
- TSStatusCode.NOT_LOGIN.getStatusCode(),
- client
- .pipeTransfer(
- PipeTransferDataNodeHandshakeV2Req.toTPipeTransferReq(
- buildHandshakeParams(null, null)))
- .getStatus()
- .getCode());
-
- final TPipeTransferResp wrongPasswordResp =
- client.pipeTransfer(
- PipeTransferDataNodeHandshakeV2Req.toTPipeTransferReq(
- buildHandshakeParams(SessionConfig.DEFAULT_USER,
"wrong-password")));
- Assert.assertNotEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
wrongPasswordResp.getStatus().getCode());
- Assert.assertEquals(
- TSStatusCode.NOT_LOGIN.getStatusCode(),
- client.pipeTransfer(buildTabletReq(1, 1)).getStatus().getCode());
-
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- client
- .pipeTransfer(
- PipeTransferDataNodeHandshakeV2Req.toTPipeTransferReq(
- buildHandshakeParams(
- SessionConfig.DEFAULT_USER,
SessionConfig.DEFAULT_PASSWORD)))
- .getStatus()
- .getCode());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- client.pipeTransfer(buildTabletReq(1, 1)).getStatus().getCode());
- }
-
- try (final IoTDBSyncClient client = createClient(dataNode)) {
- final TSOpenSessionResp openSessionResp =
client.openSession(createOpenSessionReq());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
openSessionResp.getStatus().getCode());
- try {
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- client.getBackupConfiguration().getStatus().getCode());
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- client.pipeTransfer(buildTabletReq(2, 2)).getStatus().getCode());
- } finally {
- client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
- }
- }
-
- assertRowCount(DATA_NODE_DEVICE, 2);
- }
-
- @Test
- public void testConfigNodeReceiverSessionHandling() throws Exception {
- final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
- try (final IoTDBSyncClient client = createClient(dataNode)) {
- Assert.assertEquals(
- TSStatusCode.PIPE_CONFIG_RECEIVER_HANDSHAKE_NEEDED.getStatusCode(),
-
client.pipeTransfer(buildEmptyConfigPlanReq()).getStatus().getCode());
- Assert.assertEquals(
- TSStatusCode.PIPE_HANDSHAKE_ERROR.getStatusCode(),
- client
- .pipeTransfer(
- PipeTransferConfigNodeHandshakeV1Req.toTPipeTransferReq(
-
CommonDescriptor.getInstance().getConfig().getTimestampPrecision()))
- .getStatus()
- .getCode());
- Assert.assertEquals(
- TSStatusCode.NOT_LOGIN.getStatusCode(),
- client
- .pipeTransfer(
- PipeTransferConfigNodeHandshakeV2Req.toTPipeTransferReq(
- buildHandshakeParams(null, null)))
- .getStatus()
- .getCode());
-
- final TPipeTransferResp wrongPasswordResp =
- client.pipeTransfer(
- PipeTransferConfigNodeHandshakeV2Req.toTPipeTransferReq(
- buildHandshakeParams(SessionConfig.DEFAULT_USER,
"wrong-password")));
- Assert.assertNotEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
wrongPasswordResp.getStatus().getCode());
- Assert.assertEquals(
- TSStatusCode.PIPE_CONFIG_RECEIVER_HANDSHAKE_NEEDED.getStatusCode(),
-
client.pipeTransfer(buildEmptyConfigPlanReq()).getStatus().getCode());
-
- Assert.assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- client
- .pipeTransfer(
- PipeTransferConfigNodeHandshakeV2Req.toTPipeTransferReq(
- buildHandshakeParams(
- SessionConfig.DEFAULT_USER,
SessionConfig.DEFAULT_PASSWORD)))
- .getStatus()
- .getCode());
- Assert.assertNotEquals(
- TSStatusCode.NOT_LOGIN.getStatusCode(),
-
client.pipeTransfer(buildEmptyConfigPlanReq()).getStatus().getCode());
- }
- }
-
- private TPipeTransferReq buildEmptyConfigPlanReq() {
- return new TPipeTransferReq()
- .setVersion(IoTDBSinkRequestVersion.VERSION_1.getVersion())
- .setType(PipeRequestType.TRANSFER_CONFIG_PLAN.getType())
- .setBody(ByteBuffer.allocate(0));
- }
-
- private TPipeTransferReq buildTabletReq(final long timestamp, final int
value) throws Exception {
- final Tablet tablet =
- new Tablet(
- DATA_NODE_DEVICE,
- Collections.singletonList(new MeasurementSchema("s1",
TSDataType.INT32)),
- 1);
- tablet.addTimestamp(0, timestamp);
- tablet.addValue("s1", 0, value);
- return PipeTransferTabletRawReq.toTPipeTransferReq(tablet, false);
- }
-
- private Map<String, String> buildHandshakeParams(final String username,
final String password) {
- final Map<String, String> params = new HashMap<>();
- params.put(
- PipeTransferHandshakeConstant.HANDSHAKE_KEY_CLUSTER_ID,
- "pipe-session-it-" + UUID.randomUUID());
- params.put(
- PipeTransferHandshakeConstant.HANDSHAKE_KEY_TIME_PRECISION,
- CommonDescriptor.getInstance().getConfig().getTimestampPrecision());
- if (username != null) {
- params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_USERNAME,
username);
- }
- if (password != null) {
- params.put(PipeTransferHandshakeConstant.HANDSHAKE_KEY_PASSWORD,
password);
- }
- return params;
- }
-
- private IoTDBSyncClient createClient(final DataNodeWrapper dataNode) throws
Exception {
- return new IoTDBSyncClient(
- new ThriftClientProperty.Builder().build(),
- dataNode.getIp(),
- dataNode.getPort(),
- false,
- null,
- null);
- }
-
- private TSOpenSessionReq createOpenSessionReq() {
- final TSOpenSessionReq req = new TSOpenSessionReq();
- req.setClient_protocol(TSProtocolVersion.IOTDB_SERVICE_PROTOCOL_V3);
- req.setUsername(SessionConfig.DEFAULT_USER);
- req.setPassword(SessionConfig.DEFAULT_PASSWORD);
- req.setZoneId(ZoneId.systemDefault().toString());
- req.putToConfiguration("version",
IoTDBConstant.ClientVersion.V_1_0.toString());
- req.putToConfiguration("sql_dialect", "tree");
- return req;
- }
-
- private void assertRowCount(final String device, final int expectedCount)
throws Exception {
- try (final Connection connection = EnvFactory.getEnv().getConnection();
- final Statement statement = connection.createStatement();
- final ResultSet resultSet = statement.executeQuery("SELECT COUNT(s1)
FROM " + device)) {
- Assert.assertTrue(resultSet.next());
- Assert.assertEquals(expectedCount, resultSet.getInt(1));
- }
- }
-}
