This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch cp-3c0c7fef-dev-1.3 in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 011c4c87aa06f497325bc25907f37ecd26a390da Author: Caideyipi <[email protected]> AuthorDate: Tue Aug 18 12:35:35 2026 +0800 Fix mixed line endings in dev/1.3 integration tests --- .../IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java | 96 ++-- .../commit/IoTDBRegionReconstructForIoTV1IT.java | 100 ++-- .../apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java | 566 ++++++++++----------- .../iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java | 96 ++-- .../pipe/it/autocreate/IoTDBPipeAutoDropIT.java | 2 +- 5 files changed, 430 insertions(+), 430 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java index 992e9e70d87..32302eb0e0c 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java @@ -19,16 +19,16 @@ package org.apache.iotdb.confignode.it.regionmigration.pass.commit; -import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework; -import org.apache.iotdb.confignode.rpc.thrift.TExtendRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TExtendRegionReq; import org.apache.iotdb.confignode.rpc.thrift.TShowRegionResp; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.ClusterIT; -import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.TSStatusCode; import org.awaitility.Awaitility; import org.junit.Assert; @@ -41,7 +41,7 @@ import org.slf4j.LoggerFactory; import java.sql.Connection; import java.sql.Statement; import java.util.ArrayList; -import java.util.Collections; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Optional; @@ -124,50 +124,50 @@ public class IoTDBRegionGroupExpandAndShrinkForIoTV1IT } } - @Test - public void rejectInvalidTargetDataNodeTest() throws Exception { - EnvFactory.getEnv() - .getConfig() - .getCommonConfig() - .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) - .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) - .setDataReplicationFactor(1) - .setSchemaReplicationFactor(1); - - EnvFactory.getEnv().initClusterEnvironment(1, 3); - - try (final Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection()); - final Statement statement = makeItCloseQuietly(connection.createStatement()); - SyncConfigNodeIServiceClient client = - (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { - statement.execute(INSERTION1); - statement.execute(FLUSH_COMMAND); - - Map<Integer, Set<Integer>> regionMap = getAllRegionMap(statement); - Set<Integer> allDataNodeIds = getAllDataNodes(statement); - Assert.assertFalse(regionMap.isEmpty()); - - int selectedRegion = regionMap.keySet().iterator().next(); - int unknownDataNodeId = Collections.max(allDataNodeIds) + 1000; - int configNodeId = client.showCluster().getConfigNodeList().get(0).getConfigNodeId(); - Assert.assertFalse(allDataNodeIds.contains(configNodeId)); - - assertExtendRegionRejected(client, selectedRegion, unknownDataNodeId); - assertExtendRegionRejected(client, selectedRegion, configNodeId); - Assert.assertEquals(regionMap, getAllRegionMap(statement)); - } - } - - private void assertExtendRegionRejected( - SyncConfigNodeIServiceClient client, int regionId, int dataNodeId) throws Exception { - TSStatus status = - client.extendRegion(new TExtendRegionReq(Collections.singletonList(regionId), dataNodeId)); - Assert.assertEquals(TSStatusCode.EXTEND_REGION_ERROR.getStatusCode(), status.getCode()); - Assert.assertEquals( - String.format("Target DataNode %s does not exist in the cluster", dataNodeId), - status.getMessage()); - } - + @Test + public void rejectInvalidTargetDataNodeTest() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setDataReplicationFactor(1) + .setSchemaReplicationFactor(1); + + EnvFactory.getEnv().initClusterEnvironment(1, 3); + + try (final Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection()); + final Statement statement = makeItCloseQuietly(connection.createStatement()); + SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { + statement.execute(INSERTION1); + statement.execute(FLUSH_COMMAND); + + Map<Integer, Set<Integer>> regionMap = getAllRegionMap(statement); + Set<Integer> allDataNodeIds = getAllDataNodes(statement); + Assert.assertFalse(regionMap.isEmpty()); + + int selectedRegion = regionMap.keySet().iterator().next(); + int unknownDataNodeId = Collections.max(allDataNodeIds) + 1000; + int configNodeId = client.showCluster().getConfigNodeList().get(0).getConfigNodeId(); + Assert.assertFalse(allDataNodeIds.contains(configNodeId)); + + assertExtendRegionRejected(client, selectedRegion, unknownDataNodeId); + assertExtendRegionRejected(client, selectedRegion, configNodeId); + Assert.assertEquals(regionMap, getAllRegionMap(statement)); + } + } + + private void assertExtendRegionRejected( + SyncConfigNodeIServiceClient client, int regionId, int dataNodeId) throws Exception { + TSStatus status = + client.extendRegion(new TExtendRegionReq(Collections.singletonList(regionId), dataNodeId)); + Assert.assertEquals(TSStatusCode.EXTEND_REGION_ERROR.getStatusCode(), status.getCode()); + Assert.assertEquals( + String.format("Target DataNode %s does not exist in the cluster", dataNodeId), + status.getMessage()); + } + private void regionGroupExpand( Statement statement, SyncConfigNodeIServiceClient client, diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java index fe4c337ff52..a48f5515bff 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java @@ -19,11 +19,11 @@ package org.apache.iotdb.confignode.it.regionmigration.pass.commit; -import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient; import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework; -import org.apache.iotdb.confignode.rpc.thrift.TReconstructRegionReq; +import org.apache.iotdb.confignode.rpc.thrift.TReconstructRegionReq; import org.apache.iotdb.consensus.ConsensusFactory; import org.apache.iotdb.isession.SessionDataSet; import org.apache.iotdb.it.env.EnvFactory; @@ -31,14 +31,14 @@ import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; import org.apache.iotdb.it.framework.IoTDBTestRunner; import org.apache.iotdb.itbase.category.ClusterIT; import org.apache.iotdb.rpc.StatementExecutionException; -import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.TSStatusCode; import org.apache.iotdb.session.Session; import org.apache.commons.io.FileUtils; import org.apache.tsfile.read.common.RowRecord; import org.awaitility.Awaitility; import org.junit.Assert; -import org.junit.Test; +import org.junit.Test; import org.junit.experimental.categories.Category; import org.junit.runner.RunWith; import org.slf4j.Logger; @@ -47,7 +47,7 @@ import org.slf4j.LoggerFactory; import java.io.File; import java.sql.Connection; import java.sql.Statement; -import java.util.Collections; +import java.util.Collections; import java.util.Iterator; import java.util.Map; import java.util.Set; @@ -150,49 +150,49 @@ public class IoTDBRegionReconstructForIoTV1IT extends IoTDBRegionOperationReliab Assert.assertEquals("1.0", rowRecord.getFields().get(1).getStringValue()); } } - - @Test - public void rejectInvalidTargetDataNodeTest() throws Exception { - EnvFactory.getEnv() - .getConfig() - .getCommonConfig() - .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) - .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) - .setDataReplicationFactor(1) - .setSchemaReplicationFactor(1); - - EnvFactory.getEnv().initClusterEnvironment(1, 3); - - try (Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection()); - Statement statement = makeItCloseQuietly(connection.createStatement()); - SyncConfigNodeIServiceClient client = - (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { - statement.execute(INSERTION1); - statement.execute(FLUSH_COMMAND); - - Map<Integer, Set<Integer>> regionMap = getAllRegionMap(statement); - Set<Integer> allDataNodeIds = getAllDataNodes(statement); - Assert.assertFalse(regionMap.isEmpty()); - - int selectedRegion = regionMap.keySet().iterator().next(); - int unknownDataNodeId = Collections.max(allDataNodeIds) + 1000; - int configNodeId = client.showCluster().getConfigNodeList().get(0).getConfigNodeId(); - Assert.assertFalse(allDataNodeIds.contains(configNodeId)); - - assertReconstructRegionRejected(client, selectedRegion, unknownDataNodeId); - assertReconstructRegionRejected(client, selectedRegion, configNodeId); - Assert.assertEquals(regionMap, getAllRegionMap(statement)); - } - } - - private void assertReconstructRegionRejected( - SyncConfigNodeIServiceClient client, int regionId, int dataNodeId) throws Exception { - TSStatus status = - client.reconstructRegion( - new TReconstructRegionReq(Collections.singletonList(regionId), dataNodeId)); - Assert.assertEquals(TSStatusCode.RECONSTRUCT_REGION_ERROR.getStatusCode(), status.getCode()); - Assert.assertEquals( - String.format("Target DataNode %s does not exist in the cluster", dataNodeId), - status.getMessage()); - } + + @Test + public void rejectInvalidTargetDataNodeTest() throws Exception { + EnvFactory.getEnv() + .getConfig() + .getCommonConfig() + .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS) + .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS) + .setDataReplicationFactor(1) + .setSchemaReplicationFactor(1); + + EnvFactory.getEnv().initClusterEnvironment(1, 3); + + try (Connection connection = makeItCloseQuietly(EnvFactory.getEnv().getConnection()); + Statement statement = makeItCloseQuietly(connection.createStatement()); + SyncConfigNodeIServiceClient client = + (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { + statement.execute(INSERTION1); + statement.execute(FLUSH_COMMAND); + + Map<Integer, Set<Integer>> regionMap = getAllRegionMap(statement); + Set<Integer> allDataNodeIds = getAllDataNodes(statement); + Assert.assertFalse(regionMap.isEmpty()); + + int selectedRegion = regionMap.keySet().iterator().next(); + int unknownDataNodeId = Collections.max(allDataNodeIds) + 1000; + int configNodeId = client.showCluster().getConfigNodeList().get(0).getConfigNodeId(); + Assert.assertFalse(allDataNodeIds.contains(configNodeId)); + + assertReconstructRegionRejected(client, selectedRegion, unknownDataNodeId); + assertReconstructRegionRejected(client, selectedRegion, configNodeId); + Assert.assertEquals(regionMap, getAllRegionMap(statement)); + } + } + + private void assertReconstructRegionRejected( + SyncConfigNodeIServiceClient client, int regionId, int dataNodeId) throws Exception { + TSStatus status = + client.reconstructRegion( + new TReconstructRegionReq(Collections.singletonList(regionId), dataNodeId)); + Assert.assertEquals(TSStatusCode.RECONSTRUCT_REGION_ERROR.getStatusCode(), status.getCode()); + Assert.assertEquals( + String.format("Target DataNode %s does not exist in the cluster", dataNodeId), + status.getMessage()); + } } diff --git a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java index 563cda81126..09773e4533f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBLoadTsFileAuthIT.java @@ -1,283 +1,283 @@ -/* - * 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.db.it; - -import org.apache.iotdb.commons.auth.entity.PrivilegeType; -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.ClusterIT; -import org.apache.iotdb.itbase.category.LocalStandaloneIT; -import org.apache.iotdb.jdbc.IoTDBSQLException; - -import org.apache.commons.io.FileUtils; -import org.apache.tsfile.enums.TSDataType; -import org.apache.tsfile.file.metadata.enums.TSEncoding; -import org.apache.tsfile.write.schema.MeasurementSchema; -import org.junit.After; -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.file.Files; -import java.sql.Connection; -import java.sql.ResultSet; -import java.sql.Statement; -import java.util.Collections; -import java.util.concurrent.TimeUnit; - -import static org.apache.iotdb.db.it.utils.TestUtils.assertNonQueryTestFail; -import static org.apache.iotdb.db.it.utils.TestUtils.createUser; -import static org.apache.iotdb.db.it.utils.TestUtils.executeNonQuery; -import static org.apache.iotdb.db.it.utils.TestUtils.grantUserSeriesPrivilege; - -@RunWith(IoTDBTestRunner.class) -@Category({LocalStandaloneIT.class, ClusterIT.class}) -public class IoTDBLoadTsFileAuthIT { - private static final long PARTITION_INTERVAL = 10 * 1000L; - private static final String DATABASE = "root.load_auth"; - private static final String DEVICE = DATABASE + ".d1"; - private static final MeasurementSchema MEASUREMENT = - new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE); - private static final String NO_WRITE_USER = "load_no_write_user"; - private static final String WRITE_USER = "load_write_user"; - private static final String OTHER_PATH_WRITE_USER = "load_other_path_write_user"; - private static final String ASYNC_NO_WRITE_USER = "async_load_no_write_user"; - private static final String ASYNC_WRITE_USER = "async_load_write_user"; - private static final String PASSWORD = "test123123456"; - private static final long UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES = - Long.MAX_VALUE / 4; - - private static File tmpDir; - - @BeforeClass - public static void setUp() throws Exception { - tmpDir = new File(Files.createTempDirectory("load-auth").toUri()); - EnvFactory.getEnv().getConfig().getCommonConfig().setTimePartitionInterval(PARTITION_INTERVAL); - EnvFactory.getEnv().getConfig().getCommonConfig().setAutoCreateSchemaEnabled(false); - EnvFactory.getEnv() - .getConfig() - .getDataNodeConfig() - .setMaxAllocateMemoryRatioForLoad(1.0) - .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(10 * 1024L) - .setLoadTsFileTabletConversionBatchMemorySizeInBytes( - UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES) - .setLoadActiveListeningCheckIntervalSeconds(1); - - EnvFactory.getEnv().initClusterEnvironment(); - } - - @AfterClass - public static void tearDown() throws Exception { - deleteDatabase(); - EnvFactory.getEnv().cleanClusterEnvironment(); - FileUtils.deleteDirectory(tmpDir); - } - - @After - public void cleanData() throws Exception { - deleteDatabase(); - } - - @Test - public void testLoadWithoutSchemaCheckStillChecksWriteDataPermission() throws Exception { - final File tsFile = new File(tmpDir, "1-0-0-0.tsfile"); - prepareSchemaAndTsFile(tsFile); - createUser(NO_WRITE_USER, PASSWORD); - - assertNonQueryTestFail( - String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), - "No permissions for this operation, please add privilege WRITE_DATA", - NO_WRITE_USER, - PASSWORD); - } - - @Test - public void testLoadWithoutSchemaCheckAllowsUserWithWriteDataPermission() throws Exception { - final File tsFile = new File(tmpDir, "2-0-0-0.tsfile"); - prepareSchemaAndTsFile(tsFile); - createUser(WRITE_USER, PASSWORD); - grantUserSeriesPrivilege(WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); - - executeNonQuery( - String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), - WRITE_USER, - PASSWORD); - - 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(10, resultSet.getLong(1)); - } - } - - @Test - public void testLoadWithoutSchemaCheckRejectsUserWithOtherPathWriteDataPermission() - throws Exception { - final File tsFile = new File(tmpDir, "3-0-0-0.tsfile"); - prepareSchemaAndTsFile(tsFile); - createUser(OTHER_PATH_WRITE_USER, PASSWORD); - grantUserSeriesPrivilege(OTHER_PATH_WRITE_USER, PrivilegeType.WRITE_DATA, "root.other.**"); - - assertNonQueryTestFail( - String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), - "No permissions for this operation, please add privilege WRITE_DATA", - OTHER_PATH_WRITE_USER, - PASSWORD); - } - - @Test - public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser() throws Exception { - final File noWriteTsFile = new File(tmpDir, "4-0-0-0.tsfile"); - final File writeTsFile = new File(tmpDir, "5-0-0-0.tsfile"); - prepareSchemaAndTsFile(noWriteTsFile); - generateTsFile(writeTsFile); - createUser(ASYNC_NO_WRITE_USER, PASSWORD); - createUser(ASYNC_WRITE_USER, PASSWORD); - grantUserSeriesPrivilege(ASYNC_WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); - - executeNonQuery( - String.format( - "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " - + "'verify'='false')", - noWriteTsFile.getAbsolutePath()), - ASYNC_NO_WRITE_USER, - PASSWORD); - executeNonQuery( - String.format( - "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " - + "'verify'='false')", - writeTsFile.getAbsolutePath()), - ASYNC_WRITE_USER, - PASSWORD); - - waitUntilAllActiveLoadPendingDirsAreEmpty(TimeUnit.SECONDS.toMillis(60)); - assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); - } - - private static void prepareSchemaAndTsFile(final File tsFile) throws Exception { - prepareSchema(MEASUREMENT.getType()); - generateTsFile(tsFile); - } - - private static void prepareSchema(final TSDataType dataType) throws Exception { - try (final Connection connection = EnvFactory.getEnv().getConnection(); - final Statement statement = connection.createStatement()) { - statement.execute("create database " + DATABASE); - statement.execute( - String.format( - "create timeseries %s.%s %s", DEVICE, MEASUREMENT.getMeasurementId(), dataType)); - } - } - - private static void generateTsFile(final File tsFile) throws Exception { - try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) { - generator.registerTimeseries(DEVICE, Collections.singletonList(MEASUREMENT)); - generator.generateData(DEVICE, 10, PARTITION_INTERVAL / 10, false); - } - } - - private static void deleteDatabase() throws Exception { - try (final Connection connection = EnvFactory.getEnv().getConnection(); - final Statement statement = connection.createStatement()) { - statement.execute("delete database " + DATABASE); - } catch (final IoTDBSQLException ignored) { - } - } - - private File getActiveLoadPendingDir(final DataNodeWrapper dataNodeWrapper) { - return new File( - dataNodeWrapper.getNodePath() - + File.separator - + "ext" - + File.separator - + "load" - + File.separator - + "pending"); - } - - private void waitUntilAllActiveLoadPendingDirsAreEmpty(final long timeoutMs) - throws InterruptedException { - final long deadline = System.currentTimeMillis() + timeoutMs; - while (System.currentTimeMillis() < deadline) { - boolean hasTsFile = false; - for (final DataNodeWrapper dataNodeWrapper : EnvFactory.getEnv().getDataNodeWrapperList()) { - if (containsTsFile(getActiveLoadPendingDir(dataNodeWrapper))) { - hasTsFile = true; - break; - } - } - if (!hasTsFile) { - return; - } - Thread.sleep(500L); - } - Assert.fail("Timed out waiting for active load pending dirs to become empty"); - } - - private void assertCountEventually(final long expected, final long timeoutMs) throws Exception { - final long deadline = System.currentTimeMillis() + timeoutMs; - AssertionError lastError = null; - while (System.currentTimeMillis() < deadline) { - try (final Connection connection = EnvFactory.getEnv().getConnection(); - final Statement statement = connection.createStatement(); - final ResultSet resultSet = - statement.executeQuery( - "select count(" + MEASUREMENT.getMeasurementId() + ") from " + DEVICE)) { - Assert.assertTrue(resultSet.next()); - Assert.assertEquals(expected, resultSet.getLong(1)); - return; - } catch (final AssertionError e) { - lastError = e; - } - Thread.sleep(500L); - } - if (lastError != null) { - throw lastError; - } - Assert.fail("Timed out waiting for count " + expected); - } - - private boolean containsTsFile(final File root) { - if (root == null || !root.exists()) { - return false; - } - if (root.isFile()) { - return root.getName().endsWith(".tsfile"); - } - - final File[] children = root.listFiles(); - if (children == null) { - return false; - } - for (final File child : children) { - if (containsTsFile(child)) { - return true; - } - } - return false; - } -} +/* + * 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.db.it; + +import org.apache.iotdb.commons.auth.entity.PrivilegeType; +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.ClusterIT; +import org.apache.iotdb.itbase.category.LocalStandaloneIT; +import org.apache.iotdb.jdbc.IoTDBSQLException; + +import org.apache.commons.io.FileUtils; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.enums.TSEncoding; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.After; +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.file.Files; +import java.sql.Connection; +import java.sql.ResultSet; +import java.sql.Statement; +import java.util.Collections; +import java.util.concurrent.TimeUnit; + +import static org.apache.iotdb.db.it.utils.TestUtils.assertNonQueryTestFail; +import static org.apache.iotdb.db.it.utils.TestUtils.createUser; +import static org.apache.iotdb.db.it.utils.TestUtils.executeNonQuery; +import static org.apache.iotdb.db.it.utils.TestUtils.grantUserSeriesPrivilege; + +@RunWith(IoTDBTestRunner.class) +@Category({LocalStandaloneIT.class, ClusterIT.class}) +public class IoTDBLoadTsFileAuthIT { + private static final long PARTITION_INTERVAL = 10 * 1000L; + private static final String DATABASE = "root.load_auth"; + private static final String DEVICE = DATABASE + ".d1"; + private static final MeasurementSchema MEASUREMENT = + new MeasurementSchema("s1", TSDataType.INT32, TSEncoding.RLE); + private static final String NO_WRITE_USER = "load_no_write_user"; + private static final String WRITE_USER = "load_write_user"; + private static final String OTHER_PATH_WRITE_USER = "load_other_path_write_user"; + private static final String ASYNC_NO_WRITE_USER = "async_load_no_write_user"; + private static final String ASYNC_WRITE_USER = "async_load_write_user"; + private static final String PASSWORD = "test123123456"; + private static final long UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES = + Long.MAX_VALUE / 4; + + private static File tmpDir; + + @BeforeClass + public static void setUp() throws Exception { + tmpDir = new File(Files.createTempDirectory("load-auth").toUri()); + EnvFactory.getEnv().getConfig().getCommonConfig().setTimePartitionInterval(PARTITION_INTERVAL); + EnvFactory.getEnv().getConfig().getCommonConfig().setAutoCreateSchemaEnabled(false); + EnvFactory.getEnv() + .getConfig() + .getDataNodeConfig() + .setMaxAllocateMemoryRatioForLoad(1.0) + .setLoadTsFileAnalyzeSchemaMemorySizeInBytes(10 * 1024L) + .setLoadTsFileTabletConversionBatchMemorySizeInBytes( + UNALLOCATABLE_TABLET_CONVERSION_BATCH_MEMORY_SIZE_IN_BYTES) + .setLoadActiveListeningCheckIntervalSeconds(1); + + EnvFactory.getEnv().initClusterEnvironment(); + } + + @AfterClass + public static void tearDown() throws Exception { + deleteDatabase(); + EnvFactory.getEnv().cleanClusterEnvironment(); + FileUtils.deleteDirectory(tmpDir); + } + + @After + public void cleanData() throws Exception { + deleteDatabase(); + } + + @Test + public void testLoadWithoutSchemaCheckStillChecksWriteDataPermission() throws Exception { + final File tsFile = new File(tmpDir, "1-0-0-0.tsfile"); + prepareSchemaAndTsFile(tsFile); + createUser(NO_WRITE_USER, PASSWORD); + + assertNonQueryTestFail( + String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), + "No permissions for this operation, please add privilege WRITE_DATA", + NO_WRITE_USER, + PASSWORD); + } + + @Test + public void testLoadWithoutSchemaCheckAllowsUserWithWriteDataPermission() throws Exception { + final File tsFile = new File(tmpDir, "2-0-0-0.tsfile"); + prepareSchemaAndTsFile(tsFile); + createUser(WRITE_USER, PASSWORD); + grantUserSeriesPrivilege(WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), + WRITE_USER, + PASSWORD); + + 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(10, resultSet.getLong(1)); + } + } + + @Test + public void testLoadWithoutSchemaCheckRejectsUserWithOtherPathWriteDataPermission() + throws Exception { + final File tsFile = new File(tmpDir, "3-0-0-0.tsfile"); + prepareSchemaAndTsFile(tsFile); + createUser(OTHER_PATH_WRITE_USER, PASSWORD); + grantUserSeriesPrivilege(OTHER_PATH_WRITE_USER, PrivilegeType.WRITE_DATA, "root.other.**"); + + assertNonQueryTestFail( + String.format("load \"%s\" with ('database-level'='2', 'verify'='false')", tsFile), + "No permissions for this operation, please add privilege WRITE_DATA", + OTHER_PATH_WRITE_USER, + PASSWORD); + } + + @Test + public void testAsyncLoadShouldCheckWriteDataPermissionWithStoredUser() throws Exception { + final File noWriteTsFile = new File(tmpDir, "4-0-0-0.tsfile"); + final File writeTsFile = new File(tmpDir, "5-0-0-0.tsfile"); + prepareSchemaAndTsFile(noWriteTsFile); + generateTsFile(writeTsFile); + createUser(ASYNC_NO_WRITE_USER, PASSWORD); + createUser(ASYNC_WRITE_USER, PASSWORD); + grantUserSeriesPrivilege(ASYNC_WRITE_USER, PrivilegeType.WRITE_DATA, DATABASE + ".**"); + + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " + + "'verify'='false')", + noWriteTsFile.getAbsolutePath()), + ASYNC_NO_WRITE_USER, + PASSWORD); + executeNonQuery( + String.format( + "load \"%s\" with ('database-level'='2', 'async'='true', 'on-success'='none', " + + "'verify'='false')", + writeTsFile.getAbsolutePath()), + ASYNC_WRITE_USER, + PASSWORD); + + waitUntilAllActiveLoadPendingDirsAreEmpty(TimeUnit.SECONDS.toMillis(60)); + assertCountEventually(10, TimeUnit.SECONDS.toMillis(60)); + } + + private static void prepareSchemaAndTsFile(final File tsFile) throws Exception { + prepareSchema(MEASUREMENT.getType()); + generateTsFile(tsFile); + } + + private static void prepareSchema(final TSDataType dataType) throws Exception { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("create database " + DATABASE); + statement.execute( + String.format( + "create timeseries %s.%s %s", DEVICE, MEASUREMENT.getMeasurementId(), dataType)); + } + } + + private static void generateTsFile(final File tsFile) throws Exception { + try (final TsFileGenerator generator = new TsFileGenerator(tsFile)) { + generator.registerTimeseries(DEVICE, Collections.singletonList(MEASUREMENT)); + generator.generateData(DEVICE, 10, PARTITION_INTERVAL / 10, false); + } + } + + private static void deleteDatabase() throws Exception { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement()) { + statement.execute("delete database " + DATABASE); + } catch (final IoTDBSQLException ignored) { + } + } + + private File getActiveLoadPendingDir(final DataNodeWrapper dataNodeWrapper) { + return new File( + dataNodeWrapper.getNodePath() + + File.separator + + "ext" + + File.separator + + "load" + + File.separator + + "pending"); + } + + private void waitUntilAllActiveLoadPendingDirsAreEmpty(final long timeoutMs) + throws InterruptedException { + final long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + boolean hasTsFile = false; + for (final DataNodeWrapper dataNodeWrapper : EnvFactory.getEnv().getDataNodeWrapperList()) { + if (containsTsFile(getActiveLoadPendingDir(dataNodeWrapper))) { + hasTsFile = true; + break; + } + } + if (!hasTsFile) { + return; + } + Thread.sleep(500L); + } + Assert.fail("Timed out waiting for active load pending dirs to become empty"); + } + + private void assertCountEventually(final long expected, final long timeoutMs) throws Exception { + final long deadline = System.currentTimeMillis() + timeoutMs; + AssertionError lastError = null; + while (System.currentTimeMillis() < deadline) { + try (final Connection connection = EnvFactory.getEnv().getConnection(); + final Statement statement = connection.createStatement(); + final ResultSet resultSet = + statement.executeQuery( + "select count(" + MEASUREMENT.getMeasurementId() + ") from " + DEVICE)) { + Assert.assertTrue(resultSet.next()); + Assert.assertEquals(expected, resultSet.getLong(1)); + return; + } catch (final AssertionError e) { + lastError = e; + } + Thread.sleep(500L); + } + if (lastError != null) { + throw lastError; + } + Assert.fail("Timed out waiting for count " + expected); + } + + private boolean containsTsFile(final File root) { + if (root == null || !root.exists()) { + return false; + } + if (root.isFile()) { + return root.getName().endsWith(".tsfile"); + } + + final File[] children = root.listFiles(); + if (children == null) { + return false; + } + for (final File child : children) { + if (containsTsFile(child)) { + return true; + } + } + return false; + } +} diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java index 20688c1afa2..9605c4874a4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAlterIT.java @@ -47,15 +47,15 @@ import static org.junit.Assert.fail; @Category({MultiClusterIT2AutoCreateSchema.class}) public class IoTDBPipeAlterIT extends AbstractPipeDualAutoIT { - @Override - protected void setupConfig() { - super.setupConfig(); - senderEnv - .getConfig() - .getCommonConfig() - .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(600); - } - + @Override + protected void setupConfig() { + super.setupConfig(); + senderEnv + .getConfig() + .getCommonConfig() + .setPipeHeartbeatIntervalSecondsForCollectingPipeMeta(600); + } + @Test public void testBasicAlterPipe() throws Exception { final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); @@ -588,43 +588,43 @@ public class IoTDBPipeAlterIT extends AbstractPipeDualAutoIT { "count(timeseries),", Collections.singleton("1,")); } - - @Test - public void testAlterPipeDoesNotResendCommittedData() { - final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); - - TestUtils.executeNonQueries( - senderEnv, - Arrays.asList("insert into root.db.d1(time, s1) values (1, 1), (2, 2)", "flush"), - null); - - TestUtils.executeNonQuery( - senderEnv, - String.format( - "create pipe a2b with source ('source.realtime.mode'='stream') with sink ('node-urls'='%s', 'sink.batch.enable'='false')", - receiverDataNode.getIpAndPortString()), - null); - - final Set<String> oldData = new HashSet<>(Arrays.asList("1,1.0,", "2,2.0,")); - TestUtils.assertDataEventuallyOnEnv( - receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", oldData); - - TestUtils.executeNonQuery( - receiverEnv, "delete from root.db.d1.s1 where time >= 1 and time <= 2", null); - TestUtils.assertDataEventuallyOnEnv( - receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet()); - - TestUtils.executeNonQuery( - senderEnv, "alter pipe a2b modify sink ('sink.batch.enable'='true')", null); - TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet()); - - TestUtils.executeNonQueries( - senderEnv, Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"), null); - final Set<String> newData = Collections.singleton("3,3.0,"); - TestUtils.assertDataEventuallyOnEnv( - receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData); - TestUtils.assertDataAlwaysOnEnv( - receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData); - } + + @Test + public void testAlterPipeDoesNotResendCommittedData() { + final DataNodeWrapper receiverDataNode = receiverEnv.getDataNodeWrapper(0); + + TestUtils.executeNonQueries( + senderEnv, + Arrays.asList("insert into root.db.d1(time, s1) values (1, 1), (2, 2)", "flush"), + null); + + TestUtils.executeNonQuery( + senderEnv, + String.format( + "create pipe a2b with source ('source.realtime.mode'='stream') with sink ('node-urls'='%s', 'sink.batch.enable'='false')", + receiverDataNode.getIpAndPortString()), + null); + + final Set<String> oldData = new HashSet<>(Arrays.asList("1,1.0,", "2,2.0,")); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", oldData); + + TestUtils.executeNonQuery( + receiverEnv, "delete from root.db.d1.s1 where time >= 1 and time <= 2", null); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet()); + + TestUtils.executeNonQuery( + senderEnv, "alter pipe a2b modify sink ('sink.batch.enable'='true')", null); + TestUtils.assertDataAlwaysOnEnv( + receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", Collections.emptySet()); + + TestUtils.executeNonQueries( + senderEnv, Arrays.asList("insert into root.db.d1(time, s1) values (3, 3)", "flush"), null); + final Set<String> newData = Collections.singleton("3,3.0,"); + TestUtils.assertDataEventuallyOnEnv( + receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData); + TestUtils.assertDataAlwaysOnEnv( + receiverEnv, "select * from root.db.d1", "Time,root.db.d1.s1,", newData); + } } diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java index 3f57ae849c9..8f3ca8895c7 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeAutoDropIT.java @@ -145,7 +145,7 @@ public class IoTDBPipeAutoDropIT extends AbstractPipeDualAutoIT { TestUtils.assertDataEventuallyOnEnv( senderEnv, "show pipes", - "ID,CreationTime,State,PipeSource,PipeProcessor,PipeSink,ExceptionMessage,RemainingEventCount,EstimatedRemainingSeconds,RecentFailures,", + "ID,CreationTime,State,PipeSource,PipeProcessor,PipeSink,ExceptionMessage,RemainingEventCount,EstimatedRemainingSeconds,RecentFailures,", Collections.emptySet()); } }
