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 303686e8604 Pipe: Add receiver session authentication integration
tests (#18395)
303686e8604 is described below
commit 303686e86044b0ca95e359f7f9d58afcf91d4b8e
Author: Caideyipi <[email protected]>
AuthorDate: Wed Aug 5 11:25:05 2026 +0800
Pipe: Add receiver session authentication integration tests (#18395)
---
.../auto/basic/IoTDBPipeTwoStageAggregateIT.java | 146 +++++++++++
.../IoTDBLegacyPipeReceiverAutoCreateIT.java | 159 ++++++++++++
.../single/IoTDBLegacyPipeReceiverSecurityIT.java | 148 +++++++++++
.../pipe/it/single/IoTDBPipeReceiverSessionIT.java | 281 +++++++++++++++++++++
4 files changed, 734 insertions(+)
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
new file mode 100644
index 00000000000..ac588aa91e8
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/basic/IoTDBPipeTwoStageAggregateIT.java
@@ -0,0 +1,146 @@
+/*
+ * 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/single/IoTDBLegacyPipeReceiverAutoCreateIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverAutoCreateIT.java
new file mode 100644
index 00000000000..3325c963b22
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBLegacyPipeReceiverAutoCreateIT.java
@@ -0,0 +1,159 @@
+/*
+ * 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/IoTDBLegacyPipeReceiverSecurityIT.java
index ced5190c4b6..d383ab3de8a 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/IoTDBLegacyPipeReceiverSecurityIT.java
@@ -25,11 +25,13 @@ 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.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;
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;
@@ -39,7 +41,11 @@ 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.utils.ReadWriteIOUtils;
+import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -52,11 +58,13 @@ import java.io.DataOutputStream;
import java.io.File;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.time.ZoneId;
+import java.util.Collections;
@RunWith(IoTDBTestRunner.class)
@Category({LocalStandaloneIT.class})
@@ -67,6 +75,16 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
private static final String LEGACY_DATABASE = "root.legacy_poc";
private static final String LEGACY_TIMESERIES = LEGACY_DATABASE + ".d1.s1";
+ private static final String NO_USE_PIPE_USER = "legacyNoUsePipe";
+ private static final String NO_USE_PIPE_PASSWORD = "StrngPsWd@623452";
+
+ private static final String LEGACY_TSFILE_USER = "legacyTsFileUser";
+ private static final String LEGACY_TSFILE_PASSWORD = "StrngPsWd@623453";
+ private static final String LEGACY_TSFILE_DATABASE =
"root.legacy_tsfile_auth";
+ private static final String LEGACY_TSFILE_DEVICE = LEGACY_TSFILE_DATABASE +
".d1";
+ private static final String LEGACY_TSFILE_NAME =
+ "0-" + LEGACY_TSFILE_DATABASE + "-0-0-0-0-0-0.tsfile";
+
@BeforeClass
public static void setUp() {
EnvFactory.getEnv().getConfig().getCommonConfig().setDatanodeMemoryProportion("3:3:1:1:1:0");
@@ -116,6 +134,29 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
}
}
+ @Test
+ public void testLegacyPipeRpcRequiresLoginAndUsePipePrivilege() throws
Exception {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute("CREATE USER " + NO_USE_PIPE_USER + " '" +
NO_USE_PIPE_PASSWORD + "'");
+ }
+
+ final DataNodeWrapper dataNode = EnvFactory.getEnv().getDataNodeWrapper(0);
+ try (final IoTDBSyncClient client = createClient(dataNode)) {
+ assertLegacyPipeRpcStatus(client, TSStatusCode.NOT_LOGIN);
+
+ final TSOpenSessionResp openSessionResp =
+ client.openSession(createOpenSessionReq(NO_USE_PIPE_USER,
NO_USE_PIPE_PASSWORD));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
openSessionResp.getStatus().getCode());
+ try {
+ assertLegacyPipeRpcStatus(client, TSStatusCode.NO_PERMISSION);
+ } finally {
+ client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
+ }
+ }
+ }
+
@Test
public void testLegacyPipeDataDeleteUsesAuthenticatedUserPermission() throws
Exception {
prepareLegacyPipePrivilegeEscalationData();
@@ -154,6 +195,72 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
assertLegacyPocRowCount(2);
}
+ @Test
+ public void testLegacyTsFileLoadUsesAuthenticatedUserPermission() throws
Exception {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement()) {
+ statement.execute("CREATE DATABASE " + LEGACY_TSFILE_DATABASE);
+ statement.execute(
+ "CREATE TIMESERIES " + LEGACY_TSFILE_DEVICE + ".s1 WITH
DATATYPE=INT32,ENCODING=RLE");
+ statement.execute("CREATE USER " + LEGACY_TSFILE_USER + " '" +
LEGACY_TSFILE_PASSWORD + "'");
+ statement.execute("GRANT SYSTEM ON root.** TO USER " +
LEGACY_TSFILE_USER);
+ }
+
+ final File tempDir =
Files.createTempDirectory("legacy-pipe-tsfile-auth").toFile();
+ try {
+ final File tsFile = new File(tempDir, LEGACY_TSFILE_NAME);
+ generateTsFile(tsFile, LEGACY_TSFILE_DEVICE);
+
+ final DataNodeWrapper dataNode =
EnvFactory.getEnv().getDataNodeWrapper(0);
+ try (final IoTDBSyncClient client = createClient(dataNode)) {
+ final TSOpenSessionResp openSessionResp =
+ client.openSession(createOpenSessionReq(LEGACY_TSFILE_USER,
LEGACY_TSFILE_PASSWORD));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
openSessionResp.getStatus().getCode());
+
+ try {
+ final TSStatus handshakeStatus =
+ client.handshake(
+ new TSyncIdentityInfo(
+ "legacyTsFilePrivilege",
+ System.currentTimeMillis(),
+ "UNKNOWN",
+ LEGACY_TSFILE_DATABASE));
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
handshakeStatus.getCode());
+
+ final TSStatus status = sendLegacyTsFile(client, tsFile);
+ Assert.assertEquals(TSStatusCode.PIPESERVER_ERROR.getStatusCode(),
status.getCode());
+ } finally {
+ client.closeSession(new
TSCloseSessionReq(openSessionResp.getSessionId()));
+ }
+ }
+ } finally {
+ FileUtils.deleteDirectory(tempDir);
+ }
+
+ assertTimeseriesRowCount(LEGACY_TSFILE_DEVICE, "s1", 0);
+ }
+
+ private void assertLegacyPipeRpcStatus(
+ final IoTDBSyncClient client, final TSStatusCode expectedStatusCode)
throws Exception {
+ final int expectedCode = expectedStatusCode.getStatusCode();
+ Assert.assertEquals(
+ expectedCode,
+ client
+ .handshake(
+ new TSyncIdentityInfo(
+ "legacyRpcPermission", System.currentTimeMillis(),
"UNKNOWN", ""))
+ .getCode());
+ Assert.assertEquals(
+ expectedCode,
+ client
+ .sendFile(
+ new TSyncTransportMetaInfo("permission.tsfile", 0),
ByteBuffer.wrap(new byte[] {1}))
+ .getCode());
+ Assert.assertEquals(expectedCode,
client.sendPipeData(ByteBuffer.allocate(0)).getCode());
+ }
+
private void prepareLegacyPipePrivilegeEscalationData() throws SQLException {
try (final Connection connection = EnvFactory.getEnv().getConnection();
final Statement statement = connection.createStatement()) {
@@ -202,6 +309,47 @@ public class IoTDBLegacyPipeReceiverSecurityIT {
}
}
+ private void assertTimeseriesRowCount(
+ final String device, final String measurement, final int expectedCount)
throws SQLException {
+ try (final Connection connection = EnvFactory.getEnv().getConnection();
+ final Statement statement = connection.createStatement();
+ final ResultSet resultSet =
+ statement.executeQuery("SELECT COUNT(" + measurement + ") FROM " +
device)) {
+ Assert.assertTrue(resultSet.next());
+ Assert.assertEquals(expectedCount, resultSet.getInt(1));
+ }
+ }
+
+ private void generateTsFile(final File tsFile, final String device) throws
Exception {
+ 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);
+ }
+ }
+
+ private TSStatus sendLegacyTsFile(final IoTDBSyncClient client, final File
tsFile)
+ throws Exception {
+ final TSStatus fileStatus =
+ client.sendFile(
+ new TSyncTransportMetaInfo(tsFile.getName(), 0),
+ ByteBuffer.wrap(Files.readAllBytes(tsFile.toPath())));
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
fileStatus.getCode());
+ return client.sendPipeData(
+ ByteBuffer.wrap(new TsFilePipeData("", tsFile.getName(),
1).serialize()));
+ }
+
+ 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() {
return createOpenSessionReq(SessionConfig.DEFAULT_USER,
SessionConfig.DEFAULT_PASSWORD);
}
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
new file mode 100644
index 00000000000..a1f98723e6b
--- /dev/null
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/single/IoTDBPipeReceiverSessionIT.java
@@ -0,0 +1,281 @@
+/*
+ * 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));
+ }
+ }
+}
