This is an automated email from the ASF dual-hosted git repository.
CRZbulabula 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 506f94519fb Extend region migration speed limit to transfers and
deletion (#18118)
506f94519fb is described below
commit 506f94519fb9c4545db5441df4e2f5c8a2cadc99
Author: Yongzao <[email protected]>
AuthorDate: Thu Jul 9 14:27:30 2026 +0800
Extend region migration speed limit to transfers and deletion (#18118)
---
.../iotdb/db/it/IoTDBSetConfigurationIT.java | 24 +++++++
.../apache/iotdb/consensus/iot/IoTConsensus.java | 5 +-
.../iot/snapshot/IoTConsensusRateLimiter.java | 22 ++-----
.../iotdb/consensus/pipe/IoTConsensusV2.java | 5 +-
.../consensus/ratis/RateLimitedGrpcFactory.java | 40 ++++++++++++
.../ratis/RateLimitedGrpcLogAppender.java | 76 ++++++++++++++++++++++
.../consensus/ratis/RateLimitedGrpcRpcType.java | 37 +++++++++++
.../iotdb/consensus/ratis/RatisConsensus.java | 2 +
.../iotdb/consensus/simple/SimpleConsensus.java | 5 +-
.../ratis/RateLimitedGrpcLogAppenderTest.java | 66 +++++++++++++++++++
.../java/org/apache/iotdb/db/conf/IoTDBConfig.java | 11 ++++
.../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 12 ++++
.../iotconsensusv2/IoTConsensusV2SyncSink.java | 32 ++++-----
.../schemaregion/SchemaRegionUtils.java | 14 ++++
.../schemaregion/impl/SchemaRegionMemoryImpl.java | 4 +-
.../schemaregion/impl/SchemaRegionPBTreeImpl.java | 4 +-
.../iotdb/db/storageengine/StorageEngine.java | 9 +--
.../db/storageengine/dataregion/DataRegion.java | 12 ++--
.../dataregion/wal/allocation/ElasticStrategy.java | 5 +-
.../wal/allocation/FirstCreateStrategy.java | 5 +-
.../org/apache/iotdb/db/conf/PropertiesTest.java | 30 +++++++++
.../conf/iotdb-system.properties.template | 10 ++-
.../org/apache/iotdb/commons/utils/FileUtils.java | 46 ++++++++++++-
.../RegionMigrationFileRemoveRateLimiter.java} | 40 ++++++------
.../commons/utils/RegionMigrationRateLimiter.java} | 28 +++-----
.../apache/iotdb/commons/utils/FileUtilsTest.java | 64 ++++++++++++++++++
26 files changed, 515 insertions(+), 93 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java
index e919a1c5a5f..964a67155d8 100644
---
a/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java
@@ -178,6 +178,30 @@ public class IoTDBSetConfigurationIT {
checkConfigFileContains(nodeWrapper,
"heartbeat_interval_in_ms=1000")));
}
+ @Test
+ public void testHotReloadRegionMigrationFileRemoveSpeedLimit() {
+ String key = "region_migration_file_remove_speed_limit_bytes_per_second";
+ int dataNodeId = EnvFactory.getEnv().getConfigNodeWrapperList().size();
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("set configuration \"" + key + "\"=\"65536\" on " +
dataNodeId);
+ assertAppliedConfiguration(dataNodeId, key, "65536");
+ Assert.assertTrue(
+ checkConfigFileContains(EnvFactory.getEnv().getDataNodeWrapper(0),
key + "=65536"));
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ } finally {
+ try (Connection connection = EnvFactory.getEnv().getConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("set configuration \"" + key + "\"=\"16777216\" on "
+ dataNodeId);
+ } catch (Exception e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+ Assert.assertTrue(
+ checkConfigFileContains(EnvFactory.getEnv().getDataNodeWrapper(0), key
+ "=16777216"));
+ }
+
@Test
public void testHotReloadContinuousQueryMinEveryInterval() {
try (Connection connection = EnvFactory.getEnv().getConnection();
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
index 62f5af9c86f..1e247ad599e 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java
@@ -36,6 +36,7 @@ import
org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints;
import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.IStateMachine;
@@ -378,7 +379,9 @@ public class IoTConsensus implements IConsensus {
if (!exist.get()) {
throw new ConsensusGroupNotExistException(groupId);
}
- FileUtils.deleteFileOrDirectory(new File(buildPeerDir(storageDir,
groupId)));
+ FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ new File(buildPeerDir(storageDir, groupId)),
+ RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
index bb6b95b45ad..f5607874019 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
@@ -19,22 +19,16 @@
package org.apache.iotdb.consensus.iot.snapshot;
-import com.google.common.util.concurrent.RateLimiter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter;
public class IoTConsensusRateLimiter {
- private static final Logger logger =
LoggerFactory.getLogger(IoTConsensusRateLimiter.class);
- private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE);
+ private final RegionMigrationRateLimiter rateLimiter =
RegionMigrationRateLimiter.getInstance();
private IoTConsensusRateLimiter() {}
public void init(long regionMigrationSpeedLimitBytesPerSecond) {
- rateLimiter.setRate(
- regionMigrationSpeedLimitBytesPerSecond <= 0
- ? Double.MAX_VALUE
- : regionMigrationSpeedLimitBytesPerSecond);
+ rateLimiter.init(regionMigrationSpeedLimitBytesPerSecond);
}
/**
@@ -43,15 +37,7 @@ public class IoTConsensusRateLimiter {
* @param transitDataSize the size of the data to be sent
*/
public void acquireTransitDataSizeWithRateLimiter(long transitDataSize) {
- while (transitDataSize > 0) {
- if (transitDataSize > Integer.MAX_VALUE) {
- rateLimiter.acquire(Integer.MAX_VALUE);
- transitDataSize -= Integer.MAX_VALUE;
- } else {
- rateLimiter.acquire((int) transitDataSize);
- return;
- }
- }
+ rateLimiter.acquire(transitDataSize);
}
private static final IoTConsensusRateLimiter INSTANCE = new
IoTConsensusRateLimiter();
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
index 806f499116d..cf1d4056398 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java
@@ -34,6 +34,7 @@ import
org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints;
import
org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints;
import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.IStateMachine;
@@ -323,7 +324,9 @@ public class IoTConsensusV2 implements IConsensus {
consensus.clear();
stateMachineMap.remove(groupId);
- FileUtils.deleteFileOrDirectory(new File(getPeerDir(groupId)));
+ FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ new File(getPeerDir(groupId)),
+ RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
LOGGER.info(IoTConsensusV2Messages.FINISH_DELETE_LOCAL_PEER,
CLASS_NAME, groupId);
} finally {
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcFactory.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcFactory.java
new file mode 100644
index 00000000000..1124d04b865
--- /dev/null
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcFactory.java
@@ -0,0 +1,40 @@
+/*
+ * 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.consensus.ratis;
+
+import org.apache.ratis.conf.Parameters;
+import org.apache.ratis.grpc.GrpcFactory;
+import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.leader.FollowerInfo;
+import org.apache.ratis.server.leader.LeaderState;
+import org.apache.ratis.server.leader.LogAppender;
+
+class RateLimitedGrpcFactory extends GrpcFactory {
+
+ RateLimitedGrpcFactory(Parameters parameters) {
+ super(parameters);
+ }
+
+ @Override
+ public LogAppender newLogAppender(
+ RaftServer.Division server, LeaderState leaderState, FollowerInfo
follower) {
+ return new RateLimitedGrpcLogAppender(server, leaderState, follower);
+ }
+}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppender.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppender.java
new file mode 100644
index 00000000000..41830c85dc2
--- /dev/null
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppender.java
@@ -0,0 +1,76 @@
+/*
+ * 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.consensus.ratis;
+
+import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter;
+
+import org.apache.ratis.grpc.server.GrpcLogAppender;
+import org.apache.ratis.proto.RaftProtos.FileChunkProto;
+import org.apache.ratis.proto.RaftProtos.InstallSnapshotRequestProto;
+import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.leader.FollowerInfo;
+import org.apache.ratis.server.leader.LeaderState;
+import org.apache.ratis.statemachine.SnapshotInfo;
+
+import java.util.Iterator;
+
+class RateLimitedGrpcLogAppender extends GrpcLogAppender {
+
+ private final RegionMigrationRateLimiter rateLimiter =
RegionMigrationRateLimiter.getInstance();
+
+ RateLimitedGrpcLogAppender(
+ RaftServer.Division server, LeaderState leaderState, FollowerInfo
follower) {
+ super(server, leaderState, follower);
+ }
+
+ @Override
+ public Iterable<InstallSnapshotRequestProto> newInstallSnapshotRequests(
+ String requestId, SnapshotInfo snapshot) {
+ final Iterable<InstallSnapshotRequestProto> requests =
+ super.newInstallSnapshotRequests(requestId, snapshot);
+ return () -> {
+ final Iterator<InstallSnapshotRequestProto> iterator =
requests.iterator();
+ return new Iterator<InstallSnapshotRequestProto>() {
+ @Override
+ public boolean hasNext() {
+ return iterator.hasNext();
+ }
+
+ @Override
+ public InstallSnapshotRequestProto next() {
+ final InstallSnapshotRequestProto request = iterator.next();
+ rateLimiter.acquire(getSnapshotChunkDataSize(request));
+ return request;
+ }
+ };
+ };
+ }
+
+ static long getSnapshotChunkDataSize(InstallSnapshotRequestProto request) {
+ if (!request.hasSnapshotChunk()) {
+ return 0;
+ }
+
+ return request.getSnapshotChunk().getFileChunksList().stream()
+ .map(FileChunkProto::getData)
+ .mapToLong(data -> data == null ? 0 : data.size())
+ .sum();
+ }
+}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcRpcType.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcRpcType.java
new file mode 100644
index 00000000000..391ec2faef6
--- /dev/null
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcRpcType.java
@@ -0,0 +1,37 @@
+/*
+ * 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.consensus.ratis;
+
+import org.apache.ratis.conf.Parameters;
+import org.apache.ratis.rpc.RpcFactory;
+import org.apache.ratis.rpc.RpcType;
+
+public class RateLimitedGrpcRpcType implements RpcType {
+
+ @Override
+ public String name() {
+ return RateLimitedGrpcRpcType.class.getName();
+ }
+
+ @Override
+ public RpcFactory newFactory(Parameters parameters) {
+ return new RateLimitedGrpcFactory(parameters);
+ }
+}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index b7369546e7c..4cac559646f 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -56,6 +56,7 @@ import org.apache.iotdb.consensus.ratis.utils.Utils;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
+import org.apache.ratis.RaftConfigKeys;
import org.apache.ratis.client.RaftClientRpc;
import org.apache.ratis.conf.Parameters;
import org.apache.ratis.conf.RaftProperties;
@@ -159,6 +160,7 @@ class RatisConsensus implements IConsensus {
this.storageDir = new File(config.getStorageDir());
RaftServerConfigKeys.setStorageDir(properties,
Collections.singletonList(storageDir));
+ RaftConfigKeys.Rpc.setType(properties, new RateLimitedGrpcRpcType());
GrpcConfigKeys.Server.setHost(properties,
config.getThisNodeEndPoint().getIp());
GrpcConfigKeys.Server.setPort(properties,
config.getThisNodeEndPoint().getPort());
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
index 273411a1b57..06fb39ba7e0 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.request.IConsensusRequest;
import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.consensus.IConsensus;
import org.apache.iotdb.consensus.IStateMachine;
@@ -197,7 +198,9 @@ class SimpleConsensus implements IConsensus {
(k, v) -> {
exist.set(true);
v.stop();
- FileUtils.deleteFileOrDirectory(new File(buildPeerDir(groupId)));
+ FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ new File(buildPeerDir(groupId)),
+ RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
return null;
});
if (!exist.get()) {
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppenderTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppenderTest.java
new file mode 100644
index 00000000000..8b93bdcdd1a
--- /dev/null
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppenderTest.java
@@ -0,0 +1,66 @@
+/*
+ * 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.consensus.ratis;
+
+import org.apache.ratis.RaftConfigKeys;
+import org.apache.ratis.conf.Parameters;
+import org.apache.ratis.conf.RaftProperties;
+import org.apache.ratis.proto.RaftProtos.FileChunkProto;
+import org.apache.ratis.proto.RaftProtos.InstallSnapshotRequestProto;
+import org.apache.ratis.rpc.RpcType;
+import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class RateLimitedGrpcLogAppenderTest {
+
+ @Test
+ public void testGetSnapshotChunkDataSize() {
+ final InstallSnapshotRequestProto request =
+ InstallSnapshotRequestProto.newBuilder()
+ .setSnapshotChunk(
+ InstallSnapshotRequestProto.SnapshotChunkProto.newBuilder()
+ .addFileChunks(
+ FileChunkProto.newBuilder()
+ .setData(ByteString.copyFrom(new byte[3]))
+ .build())
+ .addFileChunks(
+ FileChunkProto.newBuilder()
+ .setData(ByteString.copyFrom(new byte[5]))
+ .build()))
+ .build();
+
+ Assert.assertEquals(8,
RateLimitedGrpcLogAppender.getSnapshotChunkDataSize(request));
+ Assert.assertEquals(
+ 0,
+ RateLimitedGrpcLogAppender.getSnapshotChunkDataSize(
+ InstallSnapshotRequestProto.newBuilder().buildPartial()));
+ }
+
+ @Test
+ public void testRateLimitedGrpcRpcTypeIsResolvedByRatis() {
+ final RaftProperties properties = new RaftProperties();
+
+ RaftConfigKeys.Rpc.setType(properties, new RateLimitedGrpcRpcType());
+ final RpcType rpcType = RaftConfigKeys.Rpc.type(properties, ignored -> {});
+
+ Assert.assertTrue(rpcType instanceof RateLimitedGrpcRpcType);
+ Assert.assertTrue(rpcType.newFactory(new Parameters()) instanceof
RateLimitedGrpcFactory);
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
index ff5434c4e29..7f151e450d2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java
@@ -1103,6 +1103,7 @@ public class IoTDBConfig {
private int maxPendingBatchesNum = 5;
private double maxMemoryRatioForQueue = 0.6;
private long regionMigrationSpeedLimitBytesPerSecond = 48 * 1024 * 1024L;
+ private long regionMigrationFileRemoveSpeedLimitBytesPerSecond = 16 * 1024 *
1024L;
// Throttle the per-file snapshot-transmission progress log in IoTConsensus
to at most once per
// this interval (ms). A value <= 0 logs every file.
private long dataRegionIotSnapshotTransmissionProgressLogIntervalMs = 5000L;
@@ -1280,6 +1281,16 @@ public class IoTDBConfig {
this.regionMigrationSpeedLimitBytesPerSecond =
regionMigrationSpeedLimitBytesPerSecond;
}
+ public long getRegionMigrationFileRemoveSpeedLimitBytesPerSecond() {
+ return regionMigrationFileRemoveSpeedLimitBytesPerSecond;
+ }
+
+ public void setRegionMigrationFileRemoveSpeedLimitBytesPerSecond(
+ long regionMigrationFileRemoveSpeedLimitBytesPerSecond) {
+ this.regionMigrationFileRemoveSpeedLimitBytesPerSecond =
+ regionMigrationFileRemoveSpeedLimitBytesPerSecond;
+ }
+
public long getDataRegionIotSnapshotTransmissionProgressLogIntervalMs() {
return dataRegionIotSnapshotTransmissionProgressLogIntervalMs;
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
index 223d154c728..0830b746539 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/conf/IoTDBDescriptor.java
@@ -33,6 +33,8 @@ import org.apache.iotdb.commons.schema.SchemaConstant;
import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.utils.JVMCommonUtils;
import org.apache.iotdb.commons.utils.NodeUrlUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
+import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter;
import org.apache.iotdb.confignode.rpc.thrift.TCQConfig;
import org.apache.iotdb.confignode.rpc.thrift.TGlobalConfig;
import org.apache.iotdb.confignode.rpc.thrift.TRatisConfig;
@@ -1300,6 +1302,16 @@ public class IoTDBDescriptor {
"region_migration_speed_limit_bytes_per_second",
ConfigurationFileUtils.getConfigurationDefaultValue(
"region_migration_speed_limit_bytes_per_second"))));
+ RegionMigrationRateLimiter.getInstance()
+ .init(conf.getRegionMigrationSpeedLimitBytesPerSecond());
+ conf.setRegionMigrationFileRemoveSpeedLimitBytesPerSecond(
+ Long.parseLong(
+ properties.getProperty(
+ "region_migration_file_remove_speed_limit_bytes_per_second",
+ ConfigurationFileUtils.getConfigurationDefaultValue(
+
"region_migration_file_remove_speed_limit_bytes_per_second"))));
+ RegionMigrationFileRemoveRateLimiter.getInstance()
+ .init(conf.getRegionMigrationFileRemoveSpeedLimitBytesPerSecond());
conf.setDataRegionIotSnapshotTransmissionProgressLogIntervalMs(
Long.parseLong(
properties.getProperty(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
index 5b47b92b092..03e589e2f26 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2SyncSink.java
@@ -32,7 +32,9 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig;
import
org.apache.iotdb.commons.pipe.sink.payload.iotconsensusv2.response.IoTConsensusV2TransferFilePieceResp;
import org.apache.iotdb.commons.pipe.sink.protocol.IoTDBSink;
import org.apache.iotdb.consensus.iotconsensusv2.thrift.TCommitId;
+import
org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferReq;
import
org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferResp;
+import
org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferReq;
import
org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferResp;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics;
@@ -198,9 +200,9 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
try (final SyncIoTConsensusV2ServiceClient syncIoTConsensusV2ServiceClient
=
syncRetryClientManager.borrowClient(getFollowerUrl())) {
final TIoTConsensusV2BatchTransferResp resp;
- resp =
- syncIoTConsensusV2ServiceClient.iotConsensusV2BatchTransfer(
- tabletBatchBuilder.toTIoTConsensusV2BatchTransferReq());
+ final TIoTConsensusV2BatchTransferReq req =
+ tabletBatchBuilder.toTIoTConsensusV2BatchTransferReq();
+ resp = syncIoTConsensusV2ServiceClient.iotConsensusV2BatchTransfer(req);
final List<TSStatus> statusList =
resp.getBatchResps().stream()
@@ -260,14 +262,14 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
try (final SyncIoTConsensusV2ServiceClient syncIoTConsensusV2ServiceClient
=
syncRetryClientManager.borrowClient(getFollowerUrl())) {
progressIndex = pipeDeleteDataNodeEvent.getProgressIndex();
- resp =
- syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(
- IoTConsensusV2DeleteNodeReq.toTIoTConsensusV2TransferReq(
- pipeDeleteDataNodeEvent.getDeleteDataNode(),
- tCommitId,
- tConsensusGroupId,
- progressIndex,
- thisDataNodeId));
+ final TIoTConsensusV2TransferReq req =
+ IoTConsensusV2DeleteNodeReq.toTIoTConsensusV2TransferReq(
+ pipeDeleteDataNodeEvent.getDeleteDataNode(),
+ tCommitId,
+ tConsensusGroupId,
+ progressIndex,
+ thisDataNodeId);
+ resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req);
} catch (final Exception e) {
throw new PipeRuntimeSinkRetryTimesConfigurableException(
String.format(
@@ -332,10 +334,10 @@ public class IoTConsensusV2SyncSink extends IoTDBSink {
insertNode = pipeInsertNodeTabletInsertionEvent.getInsertNode();
progressIndex = pipeInsertNodeTabletInsertionEvent.getProgressIndex();
- resp =
- syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(
- IoTConsensusV2TabletInsertNodeReq.toTIoTConsensusV2TransferReq(
- insertNode, tCommitId, tConsensusGroupId, progressIndex,
thisDataNodeId));
+ final TIoTConsensusV2TransferReq req =
+ IoTConsensusV2TabletInsertNodeReq.toTIoTConsensusV2TransferReq(
+ insertNode, tCommitId, tConsensusGroupId, progressIndex,
thisDataNodeId);
+ resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req);
} catch (final Exception e) {
throw new PipeRuntimeSinkRetryTimesConfigurableException(
String.format(
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java
index 3b3ba14a238..df11b4b037d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/SchemaRegionUtils.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.schemaengine.schemaregion;
import org.apache.iotdb.commons.exception.MetadataException;
import org.apache.iotdb.commons.file.SystemFileFactory;
+import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.db.i18n.DataNodeSchemaMessages;
import org.slf4j.Logger;
@@ -29,6 +30,7 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Objects;
+import java.util.function.LongConsumer;
public class SchemaRegionUtils {
@@ -38,6 +40,12 @@ public class SchemaRegionUtils {
public static void deleteSchemaRegionFolder(String schemaRegionDirPath,
Logger logger)
throws MetadataException {
+ deleteSchemaRegionFolder(schemaRegionDirPath, logger, null);
+ }
+
+ public static void deleteSchemaRegionFolder(
+ String schemaRegionDirPath, Logger logger, LongConsumer
deleteRateLimiter)
+ throws MetadataException {
File schemaRegionDir =
SystemFileFactory.INSTANCE.getFile(schemaRegionDirPath);
File[] sgFiles = schemaRegionDir.listFiles();
if (sgFiles == null) {
@@ -47,6 +55,9 @@ public class SchemaRegionUtils {
}
for (File file : sgFiles) {
try {
+ if (deleteRateLimiter != null) {
+
deleteRateLimiter.accept(FileUtils.estimateFileOrDirectoryRemoveCost(file));
+ }
Files.delete(file.toPath());
logger.info(DataNodeSchemaMessages.DELETE_SCHEMA_REGION_FILE,
file.getAbsolutePath());
} catch (IOException e) {
@@ -60,6 +71,9 @@ public class SchemaRegionUtils {
}
try {
+ if (deleteRateLimiter != null) {
+
deleteRateLimiter.accept(FileUtils.estimateFileOrDirectoryRemoveCost(schemaRegionDir));
+ }
Files.delete(schemaRegionDir.toPath());
logger.info(
DataNodeSchemaMessages.DELETE_SCHEMA_REGION_FOLDER,
schemaRegionDir.getAbsolutePath());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
index 1f2408cca12..c9841112df6 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java
@@ -44,6 +44,7 @@ import org.apache.iotdb.commons.schema.view.LogicalViewSchema;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
import org.apache.iotdb.commons.utils.FileUtils;
import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -488,7 +489,8 @@ public class SchemaRegionMemoryImpl implements
ISchemaRegion {
clear();
// delete all the schema region files
- SchemaRegionUtils.deleteSchemaRegionFolder(schemaRegionDirPath, logger);
+ SchemaRegionUtils.deleteSchemaRegionFolder(
+ schemaRegionDirPath, logger,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
if
(config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
SystemInfo.getInstance()
.decreaseDirectBufferMemoryCost(config.getSchemaRatisConsensusLogAppenderBufferSizeMax());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
index 8d87ef3f061..8fc6a95bbd2 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.commons.schema.node.role.IDeviceMNode;
import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode;
import org.apache.iotdb.commons.schema.template.Template;
import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.conf.IoTDBConfig;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
@@ -491,7 +492,8 @@ public class SchemaRegionPBTreeImpl implements
ISchemaRegion {
clear();
// delete all the schema region files
- SchemaRegionUtils.deleteSchemaRegionFolder(schemaRegionDirPath, logger);
+ SchemaRegionUtils.deleteSchemaRegionFolder(
+ schemaRegionDirPath, logger,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
if
(config.getSchemaRegionConsensusProtocolClass().equals(ConsensusFactory.RATIS_CONSENSUS))
{
SystemInfo.getInstance()
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index e0a7dbdcd97..2062ba70978 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -42,6 +42,7 @@ import org.apache.iotdb.commons.schema.ttl.TTLCache;
import org.apache.iotdb.commons.service.IService;
import org.apache.iotdb.commons.service.ServiceType;
import org.apache.iotdb.commons.utils.PathUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.commons.utils.StatusUtils;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.commons.utils.TimePartitionUtils;
@@ -849,12 +850,8 @@ public class StorageEngine implements IService {
dataDir + File.separator +
IoTDBConstant.SNAPSHOT_FOLDER_NAME,
region.getDatabaseName() + FILE_NAME_SEPARATOR +
regionId.getId());
if (regionSnapshotDir.exists()) {
- try {
- FileUtils.deleteDirectory(regionSnapshotDir);
- } catch (IOException e) {
- LOGGER.error(
- StorageEngineMessages.FAILED_TO_DELETE_SNAPSHOT_DIR,
regionSnapshotDir, e);
- }
+
org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ regionSnapshotDir,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
}
}
break;
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index 9f8f11cdc50..b99cd646c2a 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -46,6 +46,7 @@ import
org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.commons.utils.RetryUtils;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.commons.utils.TimePartitionUtils;
@@ -2250,8 +2251,8 @@ public class DataRegion implements IDataRegionForQuery {
File dataRegionSystemFolder =
SystemFileFactory.INSTANCE.getFile(
systemDir + File.separator + databaseName, dataRegionIdString);
- org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParent(
- dataRegionSystemFolder);
+
org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(
+ dataRegionSystemFolder,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
} finally {
writeUnlock();
}
@@ -2341,8 +2342,8 @@ public class DataRegion implements IDataRegionForQuery {
}
} else {
if (dataRegionDataFolder.exists()) {
-
org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParent(
- dataRegionDataFolder);
+
org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(
+ dataRegionDataFolder,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
}
}
}
@@ -2385,7 +2386,8 @@ public class DataRegion implements IDataRegionForQuery {
}
} else {
if (dataRegionObjectFolder.exists()) {
-
org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectory(dataRegionObjectFolder);
+
org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ dataRegionObjectFolder,
RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
}
}
}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java
index 13bc6ebe675..23a1ae63e56 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/ElasticStrategy.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.allocation;
import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode;
@@ -88,7 +89,9 @@ public class ElasticStrategy extends
AbstractNodeAllocationStrategy {
if (walNode != null) {
walNode.close();
if (walNode.getLogDirectory().exists()) {
- FileUtils.deleteFileOrDirectory(walNode.getLogDirectory());
+ FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ walNode.getLogDirectory(),
+ RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
}
WALManager.getInstance().subtractTotalDiskUsage(walNode.getDiskUsage());
WALManager.getInstance().subtractTotalFileNum(walNode.getFileNum());
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java
index 219666773e7..f1451652c39 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/allocation/FirstCreateStrategy.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.storageengine.dataregion.wal.allocation;
import org.apache.iotdb.commons.utils.FileUtils;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager;
import org.apache.iotdb.db.storageengine.dataregion.wal.node.IWALNode;
@@ -98,7 +99,9 @@ public class FirstCreateStrategy extends
AbstractNodeAllocationStrategy {
walNode.setDeleted(true);
walNode.close();
if (walNode.getLogDirectory().exists()) {
- FileUtils.deleteFileOrDirectory(walNode.getLogDirectory());
+ FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ walNode.getLogDirectory(),
+ RegionMigrationFileRemoveRateLimiter.getInstance()::acquire);
}
WALManager.getInstance().subtractTotalDiskUsage(walNode.getDiskUsage());
WALManager.getInstance().subtractTotalFileNum(walNode.getFileNum());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
index 158692fd41b..f4daffbf303 100755
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/conf/PropertiesTest.java
@@ -20,6 +20,7 @@
package org.apache.iotdb.db.conf;
import org.apache.iotdb.commons.conf.TrimProperties;
+import org.apache.iotdb.commons.utils.RegionMigrationFileRemoveRateLimiter;
import com.tngtech.archunit.core.domain.JavaClasses;
import com.tngtech.archunit.core.importer.ClassFileImporter;
@@ -35,6 +36,35 @@ import java.util.Properties;
import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.noClasses;
public class PropertiesTest {
+ @Test
+ public void testHotReloadRegionMigrationFileRemoveSpeedLimit() throws
Exception {
+ IoTDBDescriptor descriptor = IoTDBDescriptor.getInstance();
+ long originalLimit =
+
descriptor.getConfig().getRegionMigrationFileRemoveSpeedLimitBytesPerSecond();
+ try {
+ TrimProperties properties = new TrimProperties();
+
properties.setProperty("region_migration_file_remove_speed_limit_bytes_per_second",
"65536");
+ descriptor.loadHotModifiedProps(properties);
+ Assert.assertEquals(
+ 65536,
descriptor.getConfig().getRegionMigrationFileRemoveSpeedLimitBytesPerSecond());
+ Assert.assertEquals(
+ 65536.0,
RegionMigrationFileRemoveRateLimiter.getInstance().getRate(), 0.0);
+
+
properties.setProperty("region_migration_file_remove_speed_limit_bytes_per_second",
"0");
+ descriptor.loadHotModifiedProps(properties);
+ Assert.assertEquals(
+ 0,
descriptor.getConfig().getRegionMigrationFileRemoveSpeedLimitBytesPerSecond());
+ Assert.assertEquals(
+ Double.MAX_VALUE,
RegionMigrationFileRemoveRateLimiter.getInstance().getRate(), 0.0);
+ } finally {
+ TrimProperties properties = new TrimProperties();
+ properties.setProperty(
+ "region_migration_file_remove_speed_limit_bytes_per_second",
+ Long.toString(originalLimit));
+ descriptor.loadHotModifiedProps(properties);
+ }
+ }
+
@Test
public void PropertiesWithSpace() {
IoTDBDescriptor descriptor = IoTDBDescriptor.getInstance();
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index 3e2093b53fc..6e459d6e266 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -1747,12 +1747,20 @@ data_region_iot_max_pending_batches_num = 5
# Datatype: double
data_region_iot_max_memory_ratio_for_queue = 0.6
-# The maximum transit size in byte per second for region migration
+# The maximum size in byte per second for region migration transfer.
# values less than or equal to 0 means no limit
# effectiveMode: hot_reload
# Datatype: long
region_migration_speed_limit_bytes_per_second = 50331648
+# The maximum estimated remove cost in byte per second for region migration
file cleanup.
+# Each regular file remove costs 64 KiB, and each directory remove costs 4 KiB.
+# The default value equals about 256 regular file removes per second.
+# values less than or equal to 0 means no limit
+# effectiveMode: hot_reload
+# Datatype: long
+region_migration_file_remove_speed_limit_bytes_per_second = 16777216
+
# The minimum interval (in ms) between two per-file progress logs while
transmitting a snapshot in
# IoTConsensus. A snapshot may contain a huge number of files, so logging one
line per file is
# costly; this throttles it to at most once per interval. A value <= 0 logs
every file.
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
index cc4bf13d994..dd799b0f9a7 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/FileUtils.java
@@ -48,11 +48,14 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Stack;
+import java.util.function.LongConsumer;
public class FileUtils {
private static final Logger LOGGER =
LoggerFactory.getLogger(FileUtils.class);
private static final int BUFFER_SIZE = 1024;
+ private static final long FILE_REMOVE_COST_IN_BYTES = 64 * 1024L;
+ private static final long DIRECTORY_REMOVE_COST_IN_BYTES = 4 * 1024L;
private FileUtils() {}
@@ -111,15 +114,31 @@ public class FileUtils {
}
public static void deleteFileOrDirectory(File file, boolean
quietForNoSuchFile) {
+ deleteFileOrDirectory(file, quietForNoSuchFile, null);
+ }
+
+ public static void deleteFileOrDirectoryWithRateLimiter(
+ File file, LongConsumer deleteRateLimiter) {
+ deleteFileOrDirectory(file, false, deleteRateLimiter);
+ }
+
+ public static void deleteFileOrDirectoryWithRateLimiter(
+ File file, boolean quietForNoSuchFile, LongConsumer deleteRateLimiter) {
+ deleteFileOrDirectory(file, quietForNoSuchFile, deleteRateLimiter);
+ }
+
+ private static void deleteFileOrDirectory(
+ File file, boolean quietForNoSuchFile, LongConsumer deleteRateLimiter) {
if (file.isDirectory()) {
File[] files = file.listFiles();
if (files != null) {
for (File subfile : files) {
- deleteFileOrDirectory(subfile, quietForNoSuchFile);
+ deleteFileOrDirectory(subfile, quietForNoSuchFile,
deleteRateLimiter);
}
}
}
try {
+ acquireRemovePermit(file, deleteRateLimiter);
Files.delete(file.toPath());
} catch (NoSuchFileException e) {
if (!quietForNoSuchFile) {
@@ -161,16 +180,39 @@ public class FileUtils {
}
public static void deleteDirectoryAndEmptyParent(File folder) {
- deleteFileOrDirectory(folder);
+ deleteDirectoryAndEmptyParent(folder, null);
+ }
+
+ public static void deleteDirectoryAndEmptyParentWithRateLimiter(
+ File folder, LongConsumer deleteRateLimiter) {
+ deleteDirectoryAndEmptyParent(folder, deleteRateLimiter);
+ }
+
+ private static void deleteDirectoryAndEmptyParent(File folder, LongConsumer
deleteRateLimiter) {
+ deleteFileOrDirectory(folder, false, deleteRateLimiter);
final File parentFolder = folder.getParentFile();
+ if (parentFolder == null) {
+ return;
+ }
File[] files = parentFolder.listFiles();
if (parentFolder.isDirectory() && (files == null || files.length == 0)) {
+ acquireRemovePermit(parentFolder, deleteRateLimiter);
if (!parentFolder.delete()) {
LOGGER.warn(UtilMessages.DELETE_FOLDER_FAILED,
parentFolder.getAbsolutePath());
}
}
}
+ public static long estimateFileOrDirectoryRemoveCost(File file) {
+ return file.isDirectory() ? DIRECTORY_REMOVE_COST_IN_BYTES :
FILE_REMOVE_COST_IN_BYTES;
+ }
+
+ private static void acquireRemovePermit(File file, LongConsumer
deleteRateLimiter) {
+ if (deleteRateLimiter != null && file.exists()) {
+ deleteRateLimiter.accept(estimateFileOrDirectoryRemoveCost(file));
+ }
+ }
+
public static boolean copyDir(File sourceDir, File targetDir) throws
IOException {
if (!sourceDir.exists() || !sourceDir.isDirectory()) {
LOGGER.error(UtilMessages.COPY_FOLDER_SOURCE_NOT_EXIST,
sourceDir.getAbsolutePath());
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java
similarity index 53%
copy from
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
copy to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java
index bb6b95b45ad..40f9ddf9c54 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java
@@ -17,46 +17,44 @@
* under the License.
*/
-package org.apache.iotdb.consensus.iot.snapshot;
+package org.apache.iotdb.commons.utils;
import com.google.common.util.concurrent.RateLimiter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-public class IoTConsensusRateLimiter {
- private static final Logger logger =
LoggerFactory.getLogger(IoTConsensusRateLimiter.class);
+public class RegionMigrationFileRemoveRateLimiter {
private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE);
- private IoTConsensusRateLimiter() {}
+ private RegionMigrationFileRemoveRateLimiter() {}
- public void init(long regionMigrationSpeedLimitBytesPerSecond) {
+ public void init(long regionMigrationFileRemoveSpeedLimitBytesPerSecond) {
rateLimiter.setRate(
- regionMigrationSpeedLimitBytesPerSecond <= 0
+ regionMigrationFileRemoveSpeedLimitBytesPerSecond <= 0
? Double.MAX_VALUE
- : regionMigrationSpeedLimitBytesPerSecond);
+ : regionMigrationFileRemoveSpeedLimitBytesPerSecond);
}
- /**
- * Acquire the size of the data to be sent.
- *
- * @param transitDataSize the size of the data to be sent
- */
- public void acquireTransitDataSizeWithRateLimiter(long transitDataSize) {
- while (transitDataSize > 0) {
- if (transitDataSize > Integer.MAX_VALUE) {
+ public void acquire(long estimatedRemoveCostInBytes) {
+ while (estimatedRemoveCostInBytes > 0) {
+ if (estimatedRemoveCostInBytes > Integer.MAX_VALUE) {
rateLimiter.acquire(Integer.MAX_VALUE);
- transitDataSize -= Integer.MAX_VALUE;
+ estimatedRemoveCostInBytes -= Integer.MAX_VALUE;
} else {
- rateLimiter.acquire((int) transitDataSize);
+ rateLimiter.acquire((int) estimatedRemoveCostInBytes);
return;
}
}
}
- private static final IoTConsensusRateLimiter INSTANCE = new
IoTConsensusRateLimiter();
+ @TestOnly
+ public double getRate() {
+ return rateLimiter.getRate();
+ }
+
+ private static final RegionMigrationFileRemoveRateLimiter INSTANCE =
+ new RegionMigrationFileRemoveRateLimiter();
- public static IoTConsensusRateLimiter getInstance() {
+ public static RegionMigrationFileRemoveRateLimiter getInstance() {
return INSTANCE;
}
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java
similarity index 62%
copy from
iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
copy to
iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java
index bb6b95b45ad..0e3e48f75fc 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java
@@ -17,18 +17,15 @@
* under the License.
*/
-package org.apache.iotdb.consensus.iot.snapshot;
+package org.apache.iotdb.commons.utils;
import com.google.common.util.concurrent.RateLimiter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-public class IoTConsensusRateLimiter {
- private static final Logger logger =
LoggerFactory.getLogger(IoTConsensusRateLimiter.class);
+public class RegionMigrationRateLimiter {
private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE);
- private IoTConsensusRateLimiter() {}
+ private RegionMigrationRateLimiter() {}
public void init(long regionMigrationSpeedLimitBytesPerSecond) {
rateLimiter.setRate(
@@ -37,26 +34,21 @@ public class IoTConsensusRateLimiter {
: regionMigrationSpeedLimitBytesPerSecond);
}
- /**
- * Acquire the size of the data to be sent.
- *
- * @param transitDataSize the size of the data to be sent
- */
- public void acquireTransitDataSizeWithRateLimiter(long transitDataSize) {
- while (transitDataSize > 0) {
- if (transitDataSize > Integer.MAX_VALUE) {
+ public void acquire(long sizeInBytes) {
+ while (sizeInBytes > 0) {
+ if (sizeInBytes > Integer.MAX_VALUE) {
rateLimiter.acquire(Integer.MAX_VALUE);
- transitDataSize -= Integer.MAX_VALUE;
+ sizeInBytes -= Integer.MAX_VALUE;
} else {
- rateLimiter.acquire((int) transitDataSize);
+ rateLimiter.acquire((int) sizeInBytes);
return;
}
}
}
- private static final IoTConsensusRateLimiter INSTANCE = new
IoTConsensusRateLimiter();
+ private static final RegionMigrationRateLimiter INSTANCE = new
RegionMigrationRateLimiter();
- public static IoTConsensusRateLimiter getInstance() {
+ public static RegionMigrationRateLimiter getInstance() {
return INSTANCE;
}
}
diff --git
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java
index 4c2210998b0..402634b084b 100644
---
a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java
+++
b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java
@@ -33,6 +33,8 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.util.Collections;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
public class FileUtilsTest {
private File tmpDir;
@@ -72,6 +74,68 @@ public class FileUtilsTest {
Assert.assertNull(FileUtils.getIllegalError4Directory("valid_dir"));
}
+ @Test
+ public void testDeleteFileOrDirectoryWithRateLimiter() throws IOException {
+ File deleteDir = new File(tmpDir, "deleteWithRateLimiter");
+ File subDir = new File(deleteDir, "subDir");
+ Assert.assertTrue(subDir.mkdirs());
+ Files.write(new File(deleteDir, "file1").toPath(), new byte[3]);
+ Files.write(new File(subDir, "file2").toPath(), new byte[7]);
+ Files.write(new File(subDir, "empty").toPath(), new byte[0]);
+
+ AtomicLong acquiredBytes = new AtomicLong();
+ AtomicInteger acquiredFiles = new AtomicInteger();
+ FileUtils.deleteFileOrDirectoryWithRateLimiter(
+ deleteDir,
+ removeCost -> {
+ acquiredFiles.incrementAndGet();
+ acquiredBytes.addAndGet(removeCost);
+ });
+
+ Assert.assertFalse(deleteDir.exists());
+ Assert.assertEquals(5, acquiredFiles.get());
+ Assert.assertEquals(
+ 3 * FileUtils.estimateFileOrDirectoryRemoveCost(new File("file"))
+ + 2 * FileUtils.estimateFileOrDirectoryRemoveCost(tmpDir),
+ acquiredBytes.get());
+ }
+
+ @Test
+ public void testDeleteDirectoryAndEmptyParentWithRateLimiter() throws
IOException {
+ File parentDir = new File(tmpDir, "parentDir");
+ File deleteDir = new File(parentDir, "deleteDir");
+ Assert.assertTrue(deleteDir.mkdirs());
+ Files.write(new File(deleteDir, "file").toPath(), new byte[5]);
+
+ AtomicLong acquiredBytes = new AtomicLong();
+ FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(deleteDir,
acquiredBytes::addAndGet);
+
+ Assert.assertFalse(deleteDir.exists());
+ Assert.assertFalse(parentDir.exists());
+ Assert.assertTrue(tmpDir.exists());
+ Assert.assertEquals(
+ FileUtils.estimateFileOrDirectoryRemoveCost(new File("file"))
+ + 2 * FileUtils.estimateFileOrDirectoryRemoveCost(tmpDir),
+ acquiredBytes.get());
+ }
+
+ @Test
+ public void testDeleteDirectoryAndEmptyParentWithRateLimiterAndNoParent()
throws IOException {
+ File deleteDir = new File("deleteDirWithoutParent-" + System.nanoTime());
+ try {
+ Assert.assertTrue(deleteDir.mkdirs());
+
+ AtomicLong acquiredBytes = new AtomicLong();
+ long directoryRemoveCost =
FileUtils.estimateFileOrDirectoryRemoveCost(deleteDir);
+ FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter(deleteDir,
acquiredBytes::addAndGet);
+
+ Assert.assertFalse(deleteDir.exists());
+ Assert.assertEquals(directoryRemoveCost, acquiredBytes.get());
+ } finally {
+ FileUtils.deleteFileOrDirectory(deleteDir, true);
+ }
+ }
+
private void generateFile(File tsfile) throws WriteProcessException,
IOException {
try (TsFileWriter writer = new TsFileWriter(tsfile)) {
writer.registerAlignedTimeseries(
