From 0df08a1fcaa1b6198c3416092def32a16fb84e08 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 6 Jul 2026 15:39:36 +0800 Subject: [PATCH 1/5] Limit region deletion with migration speed limit --- .../iotdb/consensus/iot/IoTConsensus.java | 5 +- .../iot/snapshot/IoTConsensusRateLimiter.java | 22 ++------ .../iotdb/consensus/pipe/IoTConsensusV2.java | 4 +- .../consensus/simple/SimpleConsensus.java | 4 +- .../apache/iotdb/db/conf/IoTDBDescriptor.java | 3 ++ .../schemaregion/SchemaRegionUtils.java | 10 ++++ .../impl/SchemaRegionMemoryImpl.java | 4 +- .../impl/SchemaRegionPBTreeImpl.java | 4 +- .../iotdb/db/storageengine/StorageEngine.java | 9 ++-- .../storageengine/dataregion/DataRegion.java | 12 +++-- .../wal/allocation/ElasticStrategy.java | 4 +- .../wal/allocation/FirstCreateStrategy.java | 4 +- .../conf/iotdb-system.properties.template | 2 +- .../apache/iotdb/commons/utils/FileUtils.java | 31 ++++++++++- .../utils/RegionMigrationRateLimiter.java | 54 +++++++++++++++++++ .../iotdb/commons/utils/FileUtilsTest.java | 41 ++++++++++++++ 16 files changed, 174 insertions(+), 39 deletions(-) create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java 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 62f5af9c86f04..f4d94131a032f 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.IoTConsensusDeleteLocalPeerKillPoints; import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints; import org.apache.iotdb.commons.utils.KillPoint.KillPoint; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.StatusUtils; import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.IStateMachine; @@ -378,7 +379,9 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException if (!exist.get()) { throw new ConsensusGroupNotExistException(groupId); } - FileUtils.deleteFileOrDirectory(new File(buildPeerDir(storageDir, groupId))); + FileUtils.deleteFileOrDirectoryWithRateLimiter( + new File(buildPeerDir(storageDir, groupId)), + RegionMigrationRateLimiter.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 bb6b95b45ad56..f5607874019b4 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 void init(long regionMigrationSpeedLimitBytesPerSecond) { * @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 806f499116d5a..448c55b5296e9 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.IoTConsensusDeleteLocalPeerKillPoints; import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints; import org.apache.iotdb.commons.utils.KillPoint.KillPoint; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.StatusUtils; import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.IStateMachine; @@ -323,7 +324,8 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException consensus.clear(); stateMachineMap.remove(groupId); - FileUtils.deleteFileOrDirectory(new File(getPeerDir(groupId))); + FileUtils.deleteFileOrDirectoryWithRateLimiter( + new File(getPeerDir(groupId)), RegionMigrationRateLimiter.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/simple/SimpleConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java index 273411a1b57a8..e84cadff07cf4 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.request.IConsensusRequest; import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.StatusUtils; import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.IStateMachine; @@ -197,7 +198,8 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException (k, v) -> { exist.set(true); v.stop(); - FileUtils.deleteFileOrDirectory(new File(buildPeerDir(groupId))); + FileUtils.deleteFileOrDirectoryWithRateLimiter( + new File(buildPeerDir(groupId)), RegionMigrationRateLimiter.getInstance()::acquire); return null; }); if (!exist.get()) { 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 ddbceab5cd7ef..b21ab0083bf07 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,7 @@ 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.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; @@ -1290,6 +1291,8 @@ private void loadIoTConsensusProps(TrimProperties properties) throws IOException "region_migration_speed_limit_bytes_per_second", ConfigurationFileUtils.getConfigurationDefaultValue( "region_migration_speed_limit_bytes_per_second")))); + RegionMigrationRateLimiter.getInstance() + .init(conf.getRegionMigrationSpeedLimitBytesPerSecond()); conf.setDataRegionIotSnapshotTransmissionProgressLogIntervalMs( Long.parseLong( properties.getProperty( 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 3b3ba14a238a7..74b5a4d6c1521 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 @@ -29,6 +29,7 @@ import java.io.IOException; import java.nio.file.Files; import java.util.Objects; +import java.util.function.LongConsumer; public class SchemaRegionUtils { @@ -38,6 +39,12 @@ private 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 +54,9 @@ public static void deleteSchemaRegionFolder(String schemaRegionDirPath, Logger l } for (File file : sgFiles) { try { + if (deleteRateLimiter != null && file.isFile()) { + deleteRateLimiter.accept(file.length()); + } Files.delete(file.toPath()); logger.info(DataNodeSchemaMessages.DELETE_SCHEMA_REGION_FILE, file.getAbsolutePath()); } catch (IOException e) { 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 1f2408cca1237..c9a8a9d16a6f5 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.viewExpression.ViewExpression; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; 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 synchronized void deleteSchemaRegion() throws MetadataException { clear(); // delete all the schema region files - SchemaRegionUtils.deleteSchemaRegionFolder(schemaRegionDirPath, logger); + SchemaRegionUtils.deleteSchemaRegionFolder( + schemaRegionDirPath, logger, RegionMigrationRateLimiter.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 8d87ef3f0618f..2093c34c97323 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.IMeasurementMNode; import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; 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 synchronized void deleteSchemaRegion() throws MetadataException { clear(); // delete all the schema region files - SchemaRegionUtils.deleteSchemaRegionFolder(schemaRegionDirPath, logger); + SchemaRegionUtils.deleteSchemaRegionFolder( + schemaRegionDirPath, logger, RegionMigrationRateLimiter.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 972ed73bd41de..b52b773c9eb3f 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.service.IService; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.StatusUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.commons.utils.TimePartitionUtils; @@ -841,12 +842,8 @@ public void deleteDataRegion(DataRegionId regionId) { 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, RegionMigrationRateLimiter.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 9f4de75f7949e..a831af1039fcc 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.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; import org.apache.iotdb.commons.utils.CommonDateTimeUtils; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.commons.utils.TestOnly; import org.apache.iotdb.commons.utils.TimePartitionUtils; @@ -2257,8 +2258,8 @@ public void deleteFolder(String systemDir) { 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, RegionMigrationRateLimiter.getInstance()::acquire); } finally { writeUnlock(); } @@ -2348,8 +2349,8 @@ private void deleteAllSGFolders(List folder) { } } else { if (dataRegionDataFolder.exists()) { - org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParent( - dataRegionDataFolder); + org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter( + dataRegionDataFolder, RegionMigrationRateLimiter.getInstance()::acquire); } } } @@ -2392,7 +2393,8 @@ private void deleteAllObjectFiles(List folders) { } } else { if (dataRegionObjectFolder.exists()) { - org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectory(dataRegionObjectFolder); + org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter( + dataRegionObjectFolder, RegionMigrationRateLimiter.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 13bc6ebe67517..9279fda58f75d 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.RegionMigrationRateLimiter; 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,8 @@ public void deleteUniqueIdAndMayDeleteWALNode(String applicantUniqueId) { if (walNode != null) { walNode.close(); if (walNode.getLogDirectory().exists()) { - FileUtils.deleteFileOrDirectory(walNode.getLogDirectory()); + FileUtils.deleteFileOrDirectoryWithRateLimiter( + walNode.getLogDirectory(), RegionMigrationRateLimiter.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 219666773e729..ba73c87b6746d 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.RegionMigrationRateLimiter; 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,8 @@ public void deleteWALNode(String applicantUniqueId) { walNode.setDeleted(true); walNode.close(); if (walNode.getLogDirectory().exists()) { - FileUtils.deleteFileOrDirectory(walNode.getLogDirectory()); + FileUtils.deleteFileOrDirectoryWithRateLimiter( + walNode.getLogDirectory(), RegionMigrationRateLimiter.getInstance()::acquire); } WALManager.getInstance().subtractTotalDiskUsage(walNode.getDiskUsage()); WALManager.getInstance().subtractTotalFileNum(walNode.getFileNum()); 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 fdf7994325540..962d5d09e0e4e 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 @@ -1730,7 +1730,7 @@ 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 data transit and region file deletion # values less than or equal to 0 means no limit # effectiveMode: hot_reload # Datatype: long 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 cc4bf13d9940f..b7a41abcfaa65 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,6 +48,7 @@ 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); @@ -111,13 +112,30 @@ public static void deleteFileOrDirectory(File file) { } 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); } } + } else if (deleteRateLimiter != null && file.isFile()) { + deleteRateLimiter.accept(file.length()); } try { Files.delete(file.toPath()); @@ -161,7 +179,16 @@ public static void deleteFileOrDirectoryWithRetry(File file) { } 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(); File[] files = parentFolder.listFiles(); if (parentFolder.isDirectory() && (files == null || files.length == 0)) { diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java new file mode 100644 index 0000000000000..0e3e48f75fc03 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java @@ -0,0 +1,54 @@ +/* + * 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.commons.utils; + +import com.google.common.util.concurrent.RateLimiter; + +public class RegionMigrationRateLimiter { + + private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE); + + private RegionMigrationRateLimiter() {} + + public void init(long regionMigrationSpeedLimitBytesPerSecond) { + rateLimiter.setRate( + regionMigrationSpeedLimitBytesPerSecond <= 0 + ? Double.MAX_VALUE + : regionMigrationSpeedLimitBytesPerSecond); + } + + public void acquire(long sizeInBytes) { + while (sizeInBytes > 0) { + if (sizeInBytes > Integer.MAX_VALUE) { + rateLimiter.acquire(Integer.MAX_VALUE); + sizeInBytes -= Integer.MAX_VALUE; + } else { + rateLimiter.acquire((int) sizeInBytes); + return; + } + } + } + + private static final RegionMigrationRateLimiter INSTANCE = new RegionMigrationRateLimiter(); + + 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 4c2210998b039..f2df776d02a7d 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.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,45 @@ public void testGetIllegalError4DirectoryRejectsEmptyPath() { 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, + fileSize -> { + acquiredFiles.incrementAndGet(); + acquiredBytes.addAndGet(fileSize); + }); + + Assert.assertFalse(deleteDir.exists()); + Assert.assertEquals(3, acquiredFiles.get()); + Assert.assertEquals(10, 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(5, acquiredBytes.get()); + } + private void generateFile(File tsfile) throws WriteProcessException, IOException { try (TsFileWriter writer = new TsFileWriter(tsfile)) { writer.registerAlignedTimeseries( From 801d4fbcab6cde41f01a408f08984bf206bf65a6 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Mon, 6 Jul 2026 16:54:13 +0800 Subject: [PATCH 2/5] Limit region migration transfer for all protocols --- .../ratis/RateLimitedGrpcFactory.java | 40 ++++++++++ .../ratis/RateLimitedGrpcLogAppender.java | 76 +++++++++++++++++++ .../ratis/RateLimitedGrpcRpcType.java | 37 +++++++++ .../iotdb/consensus/ratis/RatisConsensus.java | 2 + .../ratis/RateLimitedGrpcLogAppenderTest.java | 66 ++++++++++++++++ .../IoTConsensusV2RateLimiter.java | 58 ++++++++++++++ .../IoTConsensusV2SyncSink.java | 37 +++++---- .../IoTConsensusV2DeleteEventHandler.java | 2 + ...IoTConsensusV2TabletBatchEventHandler.java | 2 + ...onsensusV2TabletInsertionEventHandler.java | 2 + ...onsensusV2TsFileInsertionEventHandler.java | 2 + .../IoTConsensusV2RateLimiterTest.java | 62 +++++++++++++++ .../conf/iotdb-system.properties.template | 2 +- 13 files changed, 372 insertions(+), 16 deletions(-) create mode 100644 iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcFactory.java create mode 100644 iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppender.java create mode 100644 iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcRpcType.java create mode 100644 iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RateLimitedGrpcLogAppenderTest.java create mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiter.java create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java 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 0000000000000..1124d04b8651f --- /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 0000000000000..41830c85dc2f6 --- /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 newInstallSnapshotRequests( + String requestId, SnapshotInfo snapshot) { + final Iterable requests = + super.newInstallSnapshotRequests(requestId, snapshot); + return () -> { + final Iterator iterator = requests.iterator(); + return new Iterator() { + @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 0000000000000..391ec2faef608 --- /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 b7369546e7cde..4cac559646fd8 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.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 @@ public RatisConsensus(ConsensusConfig config, IStateMachine.Registry registry) { 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/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 0000000000000..8b93bdcdd1a50 --- /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/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiter.java new file mode 100644 index 0000000000000..a2a72d820c8db --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiter.java @@ -0,0 +1,58 @@ +/* + * 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.pipe.sink.protocol.iotconsensusv2; + +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; +import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferReq; +import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferReq; + +import java.nio.ByteBuffer; + +public class IoTConsensusV2RateLimiter { + + private static final RegionMigrationRateLimiter RATE_LIMITER = + RegionMigrationRateLimiter.getInstance(); + + private IoTConsensusV2RateLimiter() {} + + public static void acquire(TIoTConsensusV2TransferReq req) { + RATE_LIMITER.acquire(getTransferDataSize(req)); + } + + public static void acquire(TIoTConsensusV2BatchTransferReq req) { + RATE_LIMITER.acquire(getTransferDataSize(req)); + } + + static long getTransferDataSize(TIoTConsensusV2BatchTransferReq req) { + return req.getBatchReqs() == null + ? 0 + : req.getBatchReqs().stream() + .mapToLong(IoTConsensusV2RateLimiter::getTransferDataSize) + .sum(); + } + + static long getTransferDataSize(TIoTConsensusV2TransferReq req) { + return getRemaining(req.body); + } + + private static int getRemaining(ByteBuffer byteBuffer) { + return byteBuffer == null ? 0 : byteBuffer.duplicate().remaining(); + } +} 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 7f82ce8e9a9a0..4051f132078df 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 @@ -31,8 +31,11 @@ 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.commons.utils.RegionMigrationRateLimiter; 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; @@ -196,9 +199,10 @@ private void doTransfer() { try (final SyncIoTConsensusV2ServiceClient syncIoTConsensusV2ServiceClient = syncRetryClientManager.borrowClient(getFollowerUrl())) { final TIoTConsensusV2BatchTransferResp resp; - resp = - syncIoTConsensusV2ServiceClient.iotConsensusV2BatchTransfer( - tabletBatchBuilder.toTIoTConsensusV2BatchTransferReq()); + final TIoTConsensusV2BatchTransferReq req = + tabletBatchBuilder.toTIoTConsensusV2BatchTransferReq(); + IoTConsensusV2RateLimiter.acquire(req); + resp = syncIoTConsensusV2ServiceClient.iotConsensusV2BatchTransfer(req); final List statusList = resp.getBatchResps().stream() @@ -258,14 +262,15 @@ private void doTransfer(final PipeDeleteDataNodeEvent pipeDeleteDataNodeEvent) 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); + IoTConsensusV2RateLimiter.acquire(req); + resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req); } catch (final Exception e) { throw new PipeRuntimeSinkRetryTimesConfigurableException( String.format( @@ -330,10 +335,11 @@ private void doTransfer(PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletI 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); + IoTConsensusV2RateLimiter.acquire(req); + resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req); } catch (final Exception e) { throw new PipeRuntimeSinkRetryTimesConfigurableException( String.format( @@ -455,6 +461,7 @@ protected void transferFilePieces( : Arrays.copyOfRange(readBuffer, 0, readLength); final IoTConsensusV2TransferFilePieceResp resp; try { + RegionMigrationRateLimiter.getInstance().acquire(readLength); resp = IoTConsensusV2TransferFilePieceResp.fromTIoTConsensusV2TransferResp( syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java index 36dfdffbf4239..c018fed0ed1ea 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics; import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; +import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2RateLimiter; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; @@ -65,6 +66,7 @@ public IoTConsensusV2DeleteEventHandler( } public void transfer(AsyncIoTConsensusV2ServiceClient client) throws TException { + IoTConsensusV2RateLimiter.acquire(req); client.iotConsensusV2Transfer(req, this); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java index b107b7fe06457..ce4a870b136c0 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; +import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2RateLimiter; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.builder.IoTConsensusV2AsyncBatchReqBuilder; import org.apache.iotdb.pipe.api.event.Event; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -68,6 +69,7 @@ public IoTConsensusV2TabletBatchEventHandler( } public void transfer(final AsyncIoTConsensusV2ServiceClient client) throws TException { + IoTConsensusV2RateLimiter.acquire(req); client.iotConsensusV2BatchTransfer(req, this); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java index a74e334d5e516..2e8d5b848d3ae 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java @@ -29,6 +29,7 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; +import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2RateLimiter; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.handler.PipeTransferTabletInsertionEventHandler; import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -68,6 +69,7 @@ protected IoTConsensusV2TabletInsertionEventHandler( } public void transfer(AsyncIoTConsensusV2ServiceClient client) throws TException { + IoTConsensusV2RateLimiter.acquire(req); doTransfer(client, req); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java index 4e269aaa7e82c..d2a52b64c8834 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java @@ -25,6 +25,7 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.payload.iotconsensusv2.response.IoTConsensusV2TransferFilePieceResp; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TCommitId; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferResp; @@ -175,6 +176,7 @@ public void transfer(final AsyncIoTConsensusV2ServiceClient client) readLength == readFileBufferSize ? readBuffer : Arrays.copyOfRange(readBuffer, 0, readLength); + RegionMigrationRateLimiter.getInstance().acquire(readLength); client.iotConsensusV2Transfer( transferMod ? IoTConsensusV2TsFilePieceWithModReq.toTIoTConsensusV2TransferReq( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java new file mode 100644 index 0000000000000..ed2eb2b64bc1a --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java @@ -0,0 +1,62 @@ +/* + * 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.pipe.sink.protocol.iotconsensusv2; + +import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferReq; +import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferReq; + +import org.junit.Assert; +import org.junit.Test; + +import java.nio.ByteBuffer; +import java.util.Arrays; + +public class IoTConsensusV2RateLimiterTest { + + @Test + public void testGetTransferDataSizeFromRequestBody() { + final ByteBuffer body = ByteBuffer.wrap(new byte[8]); + body.position(2); + body.limit(6); + + final TIoTConsensusV2TransferReq req = new TIoTConsensusV2TransferReq(); + req.body = body; + + Assert.assertEquals(4, IoTConsensusV2RateLimiter.getTransferDataSize(req)); + Assert.assertEquals(2, body.position()); + Assert.assertEquals(6, body.limit()); + Assert.assertEquals( + 0, IoTConsensusV2RateLimiter.getTransferDataSize(new TIoTConsensusV2TransferReq())); + } + + @Test + public void testGetTransferDataSizeFromBatchRequest() { + final TIoTConsensusV2TransferReq req1 = new TIoTConsensusV2TransferReq(); + req1.body = ByteBuffer.wrap(new byte[3]); + final TIoTConsensusV2TransferReq req2 = new TIoTConsensusV2TransferReq(); + req2.body = ByteBuffer.wrap(new byte[5]); + + Assert.assertEquals( + 8, + IoTConsensusV2RateLimiter.getTransferDataSize( + new TIoTConsensusV2BatchTransferReq(Arrays.asList(req1, req2)))); + Assert.assertEquals( + 0, IoTConsensusV2RateLimiter.getTransferDataSize(new TIoTConsensusV2BatchTransferReq())); + } +} 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 962d5d09e0e4e..533272ce6d782 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 @@ -1730,7 +1730,7 @@ data_region_iot_max_pending_batches_num = 5 # Datatype: double data_region_iot_max_memory_ratio_for_queue = 0.6 -# The maximum size in byte per second for region migration data transit and region file deletion +# The maximum size in byte per second for region migration transfer and region file deletion # values less than or equal to 0 means no limit # effectiveMode: hot_reload # Datatype: long From e84fe3136419e0ebe98c60fe7df73b19df1cfd1d Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Wed, 8 Jul 2026 21:58:55 +0800 Subject: [PATCH 3/5] Fix region migration cleanup rate limiting --- .../iotdb/consensus/iot/IoTConsensus.java | 4 +- .../iotdb/consensus/pipe/IoTConsensusV2.java | 5 +- .../consensus/simple/SimpleConsensus.java | 5 +- .../org/apache/iotdb/db/conf/IoTDBConfig.java | 11 ++++ .../apache/iotdb/db/conf/IoTDBDescriptor.java | 9 +++ .../IoTConsensusV2RateLimiter.java | 58 ----------------- .../IoTConsensusV2SyncSink.java | 5 -- .../IoTConsensusV2DeleteEventHandler.java | 2 - ...IoTConsensusV2TabletBatchEventHandler.java | 2 - ...onsensusV2TabletInsertionEventHandler.java | 2 - ...onsensusV2TsFileInsertionEventHandler.java | 2 - .../schemaregion/SchemaRegionUtils.java | 8 ++- .../impl/SchemaRegionMemoryImpl.java | 4 +- .../impl/SchemaRegionPBTreeImpl.java | 4 +- .../iotdb/db/storageengine/StorageEngine.java | 4 +- .../storageengine/dataregion/DataRegion.java | 8 +-- .../wal/allocation/ElasticStrategy.java | 5 +- .../wal/allocation/FirstCreateStrategy.java | 5 +- .../IoTConsensusV2RateLimiterTest.java | 62 ------------------- .../conf/iotdb-system.properties.template | 9 ++- .../apache/iotdb/commons/utils/FileUtils.java | 19 +++++- .../RegionMigrationFileRemoveRateLimiter.java | 55 ++++++++++++++++ .../iotdb/commons/utils/FileUtilsTest.java | 33 ++++++++-- 23 files changed, 158 insertions(+), 163 deletions(-) delete mode 100644 iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiter.java delete mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java create mode 100644 iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java 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 f4d94131a032f..1e247ad599e1c 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,7 +36,7 @@ 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.RegionMigrationRateLimiter; +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; @@ -381,7 +381,7 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException } FileUtils.deleteFileOrDirectoryWithRateLimiter( new File(buildPeerDir(storageDir, groupId)), - RegionMigrationRateLimiter.getInstance()::acquire); + RegionMigrationFileRemoveRateLimiter.getInstance()::acquire); KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE); } 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 448c55b5296e9..cf1d405639838 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,7 +34,7 @@ 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.RegionMigrationRateLimiter; +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; @@ -325,7 +325,8 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException stateMachineMap.remove(groupId); FileUtils.deleteFileOrDirectoryWithRateLimiter( - new File(getPeerDir(groupId)), RegionMigrationRateLimiter.getInstance()::acquire); + 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/simple/SimpleConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java index e84cadff07cf4..06fb39ba7e01e 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,7 +26,7 @@ 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.RegionMigrationRateLimiter; +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; @@ -199,7 +199,8 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException exist.set(true); v.stop(); FileUtils.deleteFileOrDirectoryWithRateLimiter( - new File(buildPeerDir(groupId)), RegionMigrationRateLimiter.getInstance()::acquire); + new File(buildPeerDir(groupId)), + RegionMigrationFileRemoveRateLimiter.getInstance()::acquire); return null; }); if (!exist.get()) { 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 b0c78e496e706..5f763febc2bd2 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 @@ -1101,6 +1101,7 @@ public class IoTDBConfig { private int maxPendingBatchesNum = 5; private double maxMemoryRatioForQueue = 0.6; private long regionMigrationSpeedLimitBytesPerSecond = 48 * 1024 * 1024L; + private long regionMigrationFileRemoveSpeedLimitBytesPerSecond = 48 * 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; @@ -1278,6 +1279,16 @@ public void setRegionMigrationSpeedLimitBytesPerSecond( 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 b21ab0083bf07..bcccc7f95d727 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,7 @@ 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; @@ -1293,6 +1294,14 @@ private void loadIoTConsensusProps(TrimProperties properties) throws IOException "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/IoTConsensusV2RateLimiter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiter.java deleted file mode 100644 index a2a72d820c8db..0000000000000 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiter.java +++ /dev/null @@ -1,58 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -package org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2; - -import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; -import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferReq; -import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferReq; - -import java.nio.ByteBuffer; - -public class IoTConsensusV2RateLimiter { - - private static final RegionMigrationRateLimiter RATE_LIMITER = - RegionMigrationRateLimiter.getInstance(); - - private IoTConsensusV2RateLimiter() {} - - public static void acquire(TIoTConsensusV2TransferReq req) { - RATE_LIMITER.acquire(getTransferDataSize(req)); - } - - public static void acquire(TIoTConsensusV2BatchTransferReq req) { - RATE_LIMITER.acquire(getTransferDataSize(req)); - } - - static long getTransferDataSize(TIoTConsensusV2BatchTransferReq req) { - return req.getBatchReqs() == null - ? 0 - : req.getBatchReqs().stream() - .mapToLong(IoTConsensusV2RateLimiter::getTransferDataSize) - .sum(); - } - - static long getTransferDataSize(TIoTConsensusV2TransferReq req) { - return getRemaining(req.body); - } - - private static int getRemaining(ByteBuffer byteBuffer) { - return byteBuffer == null ? 0 : byteBuffer.duplicate().remaining(); - } -} 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 4051f132078df..b608182f13419 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 @@ -31,7 +31,6 @@ 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.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TCommitId; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferReq; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferResp; @@ -201,7 +200,6 @@ private void doTransfer() { final TIoTConsensusV2BatchTransferResp resp; final TIoTConsensusV2BatchTransferReq req = tabletBatchBuilder.toTIoTConsensusV2BatchTransferReq(); - IoTConsensusV2RateLimiter.acquire(req); resp = syncIoTConsensusV2ServiceClient.iotConsensusV2BatchTransfer(req); final List statusList = @@ -269,7 +267,6 @@ private void doTransfer(final PipeDeleteDataNodeEvent pipeDeleteDataNodeEvent) tConsensusGroupId, progressIndex, thisDataNodeId); - IoTConsensusV2RateLimiter.acquire(req); resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req); } catch (final Exception e) { throw new PipeRuntimeSinkRetryTimesConfigurableException( @@ -338,7 +335,6 @@ private void doTransfer(PipeInsertNodeTabletInsertionEvent pipeInsertNodeTabletI final TIoTConsensusV2TransferReq req = IoTConsensusV2TabletInsertNodeReq.toTIoTConsensusV2TransferReq( insertNode, tCommitId, tConsensusGroupId, progressIndex, thisDataNodeId); - IoTConsensusV2RateLimiter.acquire(req); resp = syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer(req); } catch (final Exception e) { throw new PipeRuntimeSinkRetryTimesConfigurableException( @@ -461,7 +457,6 @@ protected void transferFilePieces( : Arrays.copyOfRange(readBuffer, 0, readLength); final IoTConsensusV2TransferFilePieceResp resp; try { - RegionMigrationRateLimiter.getInstance().acquire(readLength); resp = IoTConsensusV2TransferFilePieceResp.fromTIoTConsensusV2TransferResp( syncIoTConsensusV2ServiceClient.iotConsensusV2Transfer( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java index c018fed0ed1ea..36dfdffbf4239 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2DeleteEventHandler.java @@ -29,7 +29,6 @@ import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics; import org.apache.iotdb.db.pipe.event.common.deletion.PipeDeleteDataNodeEvent; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; -import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2RateLimiter; import org.apache.iotdb.pipe.api.exception.PipeException; import org.apache.iotdb.rpc.TSStatusCode; @@ -66,7 +65,6 @@ public IoTConsensusV2DeleteEventHandler( } public void transfer(AsyncIoTConsensusV2ServiceClient client) throws TException { - IoTConsensusV2RateLimiter.acquire(req); client.iotConsensusV2Transfer(req, this); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java index ce4a870b136c0..b107b7fe06457 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletBatchEventHandler.java @@ -29,7 +29,6 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; -import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2RateLimiter; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.payload.builder.IoTConsensusV2AsyncBatchReqBuilder; import org.apache.iotdb.pipe.api.event.Event; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -69,7 +68,6 @@ public IoTConsensusV2TabletBatchEventHandler( } public void transfer(final AsyncIoTConsensusV2ServiceClient client) throws TException { - IoTConsensusV2RateLimiter.acquire(req); client.iotConsensusV2BatchTransfer(req, this); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java index 2e8d5b848d3ae..a74e334d5e516 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TabletInsertionEventHandler.java @@ -29,7 +29,6 @@ import org.apache.iotdb.db.i18n.DataNodePipeMessages; import org.apache.iotdb.db.pipe.consensus.metric.IoTConsensusV2SinkMetrics; import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2AsyncSink; -import org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2.IoTConsensusV2RateLimiter; import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.handler.PipeTransferTabletInsertionEventHandler; import org.apache.iotdb.pipe.api.event.dml.insertion.TabletInsertionEvent; import org.apache.iotdb.pipe.api.exception.PipeException; @@ -69,7 +68,6 @@ protected IoTConsensusV2TabletInsertionEventHandler( } public void transfer(AsyncIoTConsensusV2ServiceClient client) throws TException { - IoTConsensusV2RateLimiter.acquire(req); doTransfer(client, req); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java index d2a52b64c8834..4e269aaa7e82c 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/handler/IoTConsensusV2TsFileInsertionEventHandler.java @@ -25,7 +25,6 @@ import org.apache.iotdb.commons.pipe.config.PipeConfig; import org.apache.iotdb.commons.pipe.resource.log.PipeLogger; import org.apache.iotdb.commons.pipe.sink.payload.iotconsensusv2.response.IoTConsensusV2TransferFilePieceResp; -import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.RetryUtils; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TCommitId; import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferResp; @@ -176,7 +175,6 @@ public void transfer(final AsyncIoTConsensusV2ServiceClient client) readLength == readFileBufferSize ? readBuffer : Arrays.copyOfRange(readBuffer, 0, readLength); - RegionMigrationRateLimiter.getInstance().acquire(readLength); client.iotConsensusV2Transfer( transferMod ? IoTConsensusV2TsFilePieceWithModReq.toTIoTConsensusV2TransferReq( 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 74b5a4d6c1521..df11b4b037ddd 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 @@ 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; @@ -54,8 +55,8 @@ public static void deleteSchemaRegionFolder( } for (File file : sgFiles) { try { - if (deleteRateLimiter != null && file.isFile()) { - deleteRateLimiter.accept(file.length()); + if (deleteRateLimiter != null) { + deleteRateLimiter.accept(FileUtils.estimateFileOrDirectoryRemoveCost(file)); } Files.delete(file.toPath()); logger.info(DataNodeSchemaMessages.DELETE_SCHEMA_REGION_FILE, file.getAbsolutePath()); @@ -70,6 +71,9 @@ public static void deleteSchemaRegionFolder( } 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 c9a8a9d16a6f5..c9841112df648 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,7 +44,7 @@ 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.RegionMigrationRateLimiter; +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; @@ -490,7 +490,7 @@ public synchronized void deleteSchemaRegion() throws MetadataException { // delete all the schema region files SchemaRegionUtils.deleteSchemaRegionFolder( - schemaRegionDirPath, logger, RegionMigrationRateLimiter.getInstance()::acquire); + 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 2093c34c97323..8fc6a95bbd221 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,7 +32,7 @@ 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.RegionMigrationRateLimiter; +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; @@ -493,7 +493,7 @@ public synchronized void deleteSchemaRegion() throws MetadataException { // delete all the schema region files SchemaRegionUtils.deleteSchemaRegionFolder( - schemaRegionDirPath, logger, RegionMigrationRateLimiter.getInstance()::acquire); + 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 b52b773c9eb3f..b7c6d6a07f0df 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,7 +42,7 @@ 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.RegionMigrationRateLimiter; +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; @@ -843,7 +843,7 @@ public void deleteDataRegion(DataRegionId regionId) { region.getDatabaseName() + FILE_NAME_SEPARATOR + regionId.getId()); if (regionSnapshotDir.exists()) { org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter( - regionSnapshotDir, RegionMigrationRateLimiter.getInstance()::acquire); + 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 a831af1039fcc..54a3d9aa2f61c 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,7 +46,7 @@ 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.RegionMigrationRateLimiter; +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; @@ -2259,7 +2259,7 @@ public void deleteFolder(String systemDir) { SystemFileFactory.INSTANCE.getFile( systemDir + File.separator + databaseName, dataRegionIdString); org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter( - dataRegionSystemFolder, RegionMigrationRateLimiter.getInstance()::acquire); + dataRegionSystemFolder, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire); } finally { writeUnlock(); } @@ -2350,7 +2350,7 @@ private void deleteAllSGFolders(List folder) { } else { if (dataRegionDataFolder.exists()) { org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter( - dataRegionDataFolder, RegionMigrationRateLimiter.getInstance()::acquire); + dataRegionDataFolder, RegionMigrationFileRemoveRateLimiter.getInstance()::acquire); } } } @@ -2394,7 +2394,7 @@ private void deleteAllObjectFiles(List folders) { } else { if (dataRegionObjectFolder.exists()) { org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter( - dataRegionObjectFolder, RegionMigrationRateLimiter.getInstance()::acquire); + 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 9279fda58f75d..23a1ae63e56bb 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,7 +20,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.allocation; import org.apache.iotdb.commons.utils.FileUtils; -import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; +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; @@ -90,7 +90,8 @@ public void deleteUniqueIdAndMayDeleteWALNode(String applicantUniqueId) { walNode.close(); if (walNode.getLogDirectory().exists()) { FileUtils.deleteFileOrDirectoryWithRateLimiter( - walNode.getLogDirectory(), RegionMigrationRateLimiter.getInstance()::acquire); + 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 ba73c87b6746d..f1451652c39a1 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,7 +20,7 @@ package org.apache.iotdb.db.storageengine.dataregion.wal.allocation; import org.apache.iotdb.commons.utils.FileUtils; -import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; +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; @@ -100,7 +100,8 @@ public void deleteWALNode(String applicantUniqueId) { walNode.close(); if (walNode.getLogDirectory().exists()) { FileUtils.deleteFileOrDirectoryWithRateLimiter( - walNode.getLogDirectory(), RegionMigrationRateLimiter.getInstance()::acquire); + 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/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java deleted file mode 100644 index ed2eb2b64bc1a..0000000000000 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2RateLimiterTest.java +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ -package org.apache.iotdb.db.pipe.sink.protocol.iotconsensusv2; - -import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2BatchTransferReq; -import org.apache.iotdb.consensus.iotconsensusv2.thrift.TIoTConsensusV2TransferReq; - -import org.junit.Assert; -import org.junit.Test; - -import java.nio.ByteBuffer; -import java.util.Arrays; - -public class IoTConsensusV2RateLimiterTest { - - @Test - public void testGetTransferDataSizeFromRequestBody() { - final ByteBuffer body = ByteBuffer.wrap(new byte[8]); - body.position(2); - body.limit(6); - - final TIoTConsensusV2TransferReq req = new TIoTConsensusV2TransferReq(); - req.body = body; - - Assert.assertEquals(4, IoTConsensusV2RateLimiter.getTransferDataSize(req)); - Assert.assertEquals(2, body.position()); - Assert.assertEquals(6, body.limit()); - Assert.assertEquals( - 0, IoTConsensusV2RateLimiter.getTransferDataSize(new TIoTConsensusV2TransferReq())); - } - - @Test - public void testGetTransferDataSizeFromBatchRequest() { - final TIoTConsensusV2TransferReq req1 = new TIoTConsensusV2TransferReq(); - req1.body = ByteBuffer.wrap(new byte[3]); - final TIoTConsensusV2TransferReq req2 = new TIoTConsensusV2TransferReq(); - req2.body = ByteBuffer.wrap(new byte[5]); - - Assert.assertEquals( - 8, - IoTConsensusV2RateLimiter.getTransferDataSize( - new TIoTConsensusV2BatchTransferReq(Arrays.asList(req1, req2)))); - Assert.assertEquals( - 0, IoTConsensusV2RateLimiter.getTransferDataSize(new TIoTConsensusV2BatchTransferReq())); - } -} 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 533272ce6d782..9961d2da173a1 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 @@ -1730,12 +1730,19 @@ data_region_iot_max_pending_batches_num = 5 # Datatype: double data_region_iot_max_memory_ratio_for_queue = 0.6 -# The maximum size in byte per second for region migration transfer and region file deletion +# 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. +# values less than or equal to 0 means no limit +# effectiveMode: hot_reload +# Datatype: long +region_migration_file_remove_speed_limit_bytes_per_second = 50331648 + # 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 b7a41abcfaa65..dd799b0f9a7fa 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 @@ -54,6 +54,8 @@ 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() {} @@ -134,10 +136,9 @@ private static void deleteFileOrDirectory( deleteFileOrDirectory(subfile, quietForNoSuchFile, deleteRateLimiter); } } - } else if (deleteRateLimiter != null && file.isFile()) { - deleteRateLimiter.accept(file.length()); } try { + acquireRemovePermit(file, deleteRateLimiter); Files.delete(file.toPath()); } catch (NoSuchFileException e) { if (!quietForNoSuchFile) { @@ -190,14 +191,28 @@ public static void deleteDirectoryAndEmptyParentWithRateLimiter( 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/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java new file mode 100644 index 0000000000000..1663b0e0326e4 --- /dev/null +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java @@ -0,0 +1,55 @@ +/* + * 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.commons.utils; + +import com.google.common.util.concurrent.RateLimiter; + +public class RegionMigrationFileRemoveRateLimiter { + + private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE); + + private RegionMigrationFileRemoveRateLimiter() {} + + public void init(long regionMigrationFileRemoveSpeedLimitBytesPerSecond) { + rateLimiter.setRate( + regionMigrationFileRemoveSpeedLimitBytesPerSecond <= 0 + ? Double.MAX_VALUE + : regionMigrationFileRemoveSpeedLimitBytesPerSecond); + } + + public void acquire(long estimatedRemoveCostInBytes) { + while (estimatedRemoveCostInBytes > 0) { + if (estimatedRemoveCostInBytes > Integer.MAX_VALUE) { + rateLimiter.acquire(Integer.MAX_VALUE); + estimatedRemoveCostInBytes -= Integer.MAX_VALUE; + } else { + rateLimiter.acquire((int) estimatedRemoveCostInBytes); + return; + } + } + } + + private static final RegionMigrationFileRemoveRateLimiter INSTANCE = + new RegionMigrationFileRemoveRateLimiter(); + + public static RegionMigrationFileRemoveRateLimiter 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 f2df776d02a7d..402634b084bde 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 @@ -87,14 +87,17 @@ public void testDeleteFileOrDirectoryWithRateLimiter() throws IOException { AtomicInteger acquiredFiles = new AtomicInteger(); FileUtils.deleteFileOrDirectoryWithRateLimiter( deleteDir, - fileSize -> { + removeCost -> { acquiredFiles.incrementAndGet(); - acquiredBytes.addAndGet(fileSize); + acquiredBytes.addAndGet(removeCost); }); Assert.assertFalse(deleteDir.exists()); - Assert.assertEquals(3, acquiredFiles.get()); - Assert.assertEquals(10, acquiredBytes.get()); + Assert.assertEquals(5, acquiredFiles.get()); + Assert.assertEquals( + 3 * FileUtils.estimateFileOrDirectoryRemoveCost(new File("file")) + + 2 * FileUtils.estimateFileOrDirectoryRemoveCost(tmpDir), + acquiredBytes.get()); } @Test @@ -110,7 +113,27 @@ public void testDeleteDirectoryAndEmptyParentWithRateLimiter() throws IOExceptio Assert.assertFalse(deleteDir.exists()); Assert.assertFalse(parentDir.exists()); Assert.assertTrue(tmpDir.exists()); - Assert.assertEquals(5, acquiredBytes.get()); + 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 { From e89a1a9224305c41b48b131df340dd12c5ad6c00 Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Wed, 8 Jul 2026 23:04:59 +0800 Subject: [PATCH 4/5] Add tests for region migration cleanup limit --- .../iotdb/db/it/IoTDBSetConfigurationIT.java | 24 +++++++++++++++ .../apache/iotdb/db/conf/PropertiesTest.java | 30 +++++++++++++++++++ .../RegionMigrationFileRemoveRateLimiter.java | 5 ++++ 3 files changed, 59 insertions(+) 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 e919a1c5a5f2d..e20efbfb4df78 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 void testRejectedHotReloadDoesNotUpdateAppliedConfiguration() { 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 + "\"=\"50331648\" on " + dataNodeId); + } catch (Exception e) { + Assert.fail(e.getMessage()); + } + } + Assert.assertTrue( + checkConfigFileContains(EnvFactory.getEnv().getDataNodeWrapper(0), key + "=50331648")); + } + @Test public void testHotReloadContinuousQueryMinEveryInterval() { try (Connection connection = EnvFactory.getEnv().getConnection(); 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 158692fd41bfc..f4daffbf303b9 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 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/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java index 1663b0e0326e4..40f9ddf9c5448 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationFileRemoveRateLimiter.java @@ -46,6 +46,11 @@ public void acquire(long estimatedRemoveCostInBytes) { } } + @TestOnly + public double getRate() { + return rateLimiter.getRate(); + } + private static final RegionMigrationFileRemoveRateLimiter INSTANCE = new RegionMigrationFileRemoveRateLimiter(); From 7eeb8b175704aa453ae639c58081f18aca616ffc Mon Sep 17 00:00:00 2001 From: Yongzao <532741407@qq.com> Date: Thu, 9 Jul 2026 12:43:43 +0800 Subject: [PATCH 5/5] Adjust region migration cleanup limit default --- .../java/org/apache/iotdb/db/it/IoTDBSetConfigurationIT.java | 4 ++-- .../src/main/java/org/apache/iotdb/db/conf/IoTDBConfig.java | 2 +- .../assembly/resources/conf/iotdb-system.properties.template | 3 ++- 3 files changed, 5 insertions(+), 4 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 e20efbfb4df78..964a67155d881 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 @@ -193,13 +193,13 @@ public void testHotReloadRegionMigrationFileRemoveSpeedLimit() { } finally { try (Connection connection = EnvFactory.getEnv().getConnection(); Statement statement = connection.createStatement()) { - statement.execute("set configuration \"" + key + "\"=\"50331648\" on " + dataNodeId); + statement.execute("set configuration \"" + key + "\"=\"16777216\" on " + dataNodeId); } catch (Exception e) { Assert.fail(e.getMessage()); } } Assert.assertTrue( - checkConfigFileContains(EnvFactory.getEnv().getDataNodeWrapper(0), key + "=50331648")); + checkConfigFileContains(EnvFactory.getEnv().getDataNodeWrapper(0), key + "=16777216")); } @Test 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 5f763febc2bd2..fe74770d609ff 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 @@ -1101,7 +1101,7 @@ public class IoTDBConfig { private int maxPendingBatchesNum = 5; private double maxMemoryRatioForQueue = 0.6; private long regionMigrationSpeedLimitBytesPerSecond = 48 * 1024 * 1024L; - private long regionMigrationFileRemoveSpeedLimitBytesPerSecond = 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; 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 9961d2da173a1..8d71f072db1bc 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 @@ -1738,10 +1738,11 @@ 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 = 50331648 +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