This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch test/pipe-receiver-session-it in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 9eab84fb250fb950df60d49aee54e751d0f50d36 Author: Caideyipi <[email protected]> AuthorDate: Tue Aug 4 19:24:04 2026 +0800 Pipe: Add receiver session authentication integration tests --- .../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)); + } + } +}
