This is an automated email from the ASF dual-hosted git repository. CRZbulabula pushed a commit to branch feat-region-delete-speed-limit in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 0df08a1fcaa1b6198c3416092def32a16fb84e08 Author: Yongzao <[email protected]> AuthorDate: Mon Jul 6 15:39:36 2026 +0800 Limit region deletion with migration speed limit --- .../apache/iotdb/consensus/iot/IoTConsensus.java | 5 ++- .../iot/snapshot/IoTConsensusRateLimiter.java | 22 +++--------- .../iotdb/consensus/pipe/IoTConsensusV2.java | 4 ++- .../iotdb/consensus/simple/SimpleConsensus.java | 4 ++- .../org/apache/iotdb/db/conf/IoTDBDescriptor.java | 3 ++ .../schemaregion/SchemaRegionUtils.java | 10 ++++++ .../schemaregion/impl/SchemaRegionMemoryImpl.java | 4 ++- .../schemaregion/impl/SchemaRegionPBTreeImpl.java | 4 ++- .../iotdb/db/storageengine/StorageEngine.java | 9 ++--- .../db/storageengine/dataregion/DataRegion.java | 12 ++++--- .../dataregion/wal/allocation/ElasticStrategy.java | 4 ++- .../wal/allocation/FirstCreateStrategy.java | 4 ++- .../conf/iotdb-system.properties.template | 2 +- .../org/apache/iotdb/commons/utils/FileUtils.java | 31 ++++++++++++++-- .../commons/utils/RegionMigrationRateLimiter.java} | 28 ++++++--------- .../apache/iotdb/commons/utils/FileUtilsTest.java | 41 ++++++++++++++++++++++ 16 files changed, 130 insertions(+), 57 deletions(-) diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java index 62f5af9c86f..f4d94131a03 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java @@ -36,6 +36,7 @@ import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints; import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints; import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints; import org.apache.iotdb.commons.utils.KillPoint.KillPoint; +import org.apache.iotdb.commons.utils.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 class IoTConsensus implements IConsensus { if (!exist.get()) { throw new ConsensusGroupNotExistException(groupId); } - FileUtils.deleteFileOrDirectory(new File(buildPeerDir(storageDir, groupId))); + FileUtils.deleteFileOrDirectoryWithRateLimiter( + new File(buildPeerDir(storageDir, groupId)), + 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 bb6b95b45ad..f5607874019 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java @@ -19,22 +19,16 @@ package org.apache.iotdb.consensus.iot.snapshot; -import com.google.common.util.concurrent.RateLimiter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; public class IoTConsensusRateLimiter { - private static final Logger logger = LoggerFactory.getLogger(IoTConsensusRateLimiter.class); - private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE); + private final RegionMigrationRateLimiter rateLimiter = RegionMigrationRateLimiter.getInstance(); private IoTConsensusRateLimiter() {} public void init(long regionMigrationSpeedLimitBytesPerSecond) { - rateLimiter.setRate( - regionMigrationSpeedLimitBytesPerSecond <= 0 - ? Double.MAX_VALUE - : regionMigrationSpeedLimitBytesPerSecond); + rateLimiter.init(regionMigrationSpeedLimitBytesPerSecond); } /** @@ -43,15 +37,7 @@ public class IoTConsensusRateLimiter { * @param transitDataSize the size of the data to be sent */ public void acquireTransitDataSizeWithRateLimiter(long transitDataSize) { - while (transitDataSize > 0) { - if (transitDataSize > Integer.MAX_VALUE) { - rateLimiter.acquire(Integer.MAX_VALUE); - transitDataSize -= Integer.MAX_VALUE; - } else { - rateLimiter.acquire((int) transitDataSize); - return; - } - } + rateLimiter.acquire(transitDataSize); } private static final IoTConsensusRateLimiter INSTANCE = new IoTConsensusRateLimiter(); diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java index 806f499116d..448c55b5296 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/pipe/IoTConsensusV2.java @@ -34,6 +34,7 @@ import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints; import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusDeleteLocalPeerKillPoints; import org.apache.iotdb.commons.utils.KillPoint.IoTConsensusRemovePeerCoordinatorKillPoints; import org.apache.iotdb.commons.utils.KillPoint.KillPoint; +import org.apache.iotdb.commons.utils.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 class IoTConsensusV2 implements IConsensus { 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 273411a1b57..e84cadff07c 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/simple/SimpleConsensus.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics; import org.apache.iotdb.commons.utils.FileUtils; +import org.apache.iotdb.commons.utils.RegionMigrationRateLimiter; import org.apache.iotdb.commons.utils.StatusUtils; import org.apache.iotdb.consensus.IConsensus; import org.apache.iotdb.consensus.IStateMachine; @@ -197,7 +198,8 @@ class SimpleConsensus implements IConsensus { (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 ddbceab5cd7..b21ab0083bf 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.schema.SchemaConstant; import org.apache.iotdb.commons.service.metric.MetricService; import org.apache.iotdb.commons.utils.JVMCommonUtils; import org.apache.iotdb.commons.utils.NodeUrlUtils; +import org.apache.iotdb.commons.utils.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 @@ public class IoTDBDescriptor { "region_migration_speed_limit_bytes_per_second", ConfigurationFileUtils.getConfigurationDefaultValue( "region_migration_speed_limit_bytes_per_second")))); + RegionMigrationRateLimiter.getInstance() + .init(conf.getRegionMigrationSpeedLimitBytesPerSecond()); conf.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 3b3ba14a238..74b5a4d6c15 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.File; 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 @@ public class SchemaRegionUtils { public static void deleteSchemaRegionFolder(String schemaRegionDirPath, Logger logger) throws MetadataException { + deleteSchemaRegionFolder(schemaRegionDirPath, logger, null); + } + + public static void deleteSchemaRegionFolder( + String schemaRegionDirPath, Logger logger, LongConsumer deleteRateLimiter) + throws MetadataException { File schemaRegionDir = SystemFileFactory.INSTANCE.getFile(schemaRegionDirPath); File[] sgFiles = schemaRegionDir.listFiles(); if (sgFiles == null) { @@ -47,6 +54,9 @@ public class SchemaRegionUtils { } 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 1f2408cca12..c9a8a9d16a6 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionMemoryImpl.java @@ -44,6 +44,7 @@ import org.apache.iotdb.commons.schema.view.LogicalViewSchema; import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression; import org.apache.iotdb.commons.utils.FileUtils; import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.commons.utils.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 class SchemaRegionMemoryImpl implements ISchemaRegion { 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 8d87ef3f061..2093c34c973 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/schemaengine/schemaregion/impl/SchemaRegionPBTreeImpl.java @@ -32,6 +32,7 @@ import org.apache.iotdb.commons.schema.node.role.IDeviceMNode; import org.apache.iotdb.commons.schema.node.role.IMeasurementMNode; import org.apache.iotdb.commons.schema.template.Template; import org.apache.iotdb.commons.schema.view.viewExpression.ViewExpression; +import org.apache.iotdb.commons.utils.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 class SchemaRegionPBTreeImpl implements ISchemaRegion { 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 972ed73bd41..b52b773c9eb 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -42,6 +42,7 @@ import org.apache.iotdb.commons.schema.ttl.TTLCache; import org.apache.iotdb.commons.service.IService; import org.apache.iotdb.commons.service.ServiceType; import org.apache.iotdb.commons.utils.PathUtils; +import org.apache.iotdb.commons.utils.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 class StorageEngine implements IService { dataDir + File.separator + IoTDBConstant.SNAPSHOT_FOLDER_NAME, region.getDatabaseName() + FILE_NAME_SEPARATOR + regionId.getId()); if (regionSnapshotDir.exists()) { - try { - FileUtils.deleteDirectory(regionSnapshotDir); - } catch (IOException e) { - LOGGER.error( - StorageEngineMessages.FAILED_TO_DELETE_SNAPSHOT_DIR, regionSnapshotDir, e); - } + org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter( + regionSnapshotDir, 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 9f4de75f794..a831af1039f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -46,6 +46,7 @@ import org.apache.iotdb.commons.service.metric.PerformanceOverviewMetrics; import org.apache.iotdb.commons.service.metric.enums.Metric; import org.apache.iotdb.commons.service.metric.enums.Tag; import org.apache.iotdb.commons.utils.CommonDateTimeUtils; +import org.apache.iotdb.commons.utils.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 class DataRegion implements IDataRegionForQuery { File dataRegionSystemFolder = SystemFileFactory.INSTANCE.getFile( systemDir + File.separator + databaseName, dataRegionIdString); - org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParent( - dataRegionSystemFolder); + org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter( + dataRegionSystemFolder, RegionMigrationRateLimiter.getInstance()::acquire); } finally { writeUnlock(); } @@ -2348,8 +2349,8 @@ public class DataRegion implements IDataRegionForQuery { } } else { if (dataRegionDataFolder.exists()) { - org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParent( - dataRegionDataFolder); + org.apache.iotdb.commons.utils.FileUtils.deleteDirectoryAndEmptyParentWithRateLimiter( + dataRegionDataFolder, RegionMigrationRateLimiter.getInstance()::acquire); } } } @@ -2392,7 +2393,8 @@ public class DataRegion implements IDataRegionForQuery { } } else { if (dataRegionObjectFolder.exists()) { - org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectory(dataRegionObjectFolder); + org.apache.iotdb.commons.utils.FileUtils.deleteFileOrDirectoryWithRateLimiter( + dataRegionObjectFolder, 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 13bc6ebe675..9279fda58f7 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 class ElasticStrategy extends AbstractNodeAllocationStrategy { 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 219666773e7..ba73c87b674 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 class FirstCreateStrategy extends AbstractNodeAllocationStrategy { 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 fdf79943255..962d5d09e0e 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 cc4bf13d994..b7a41abcfaa 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.ArrayList; import java.util.Arrays; import java.util.List; import java.util.Stack; +import java.util.function.LongConsumer; public class FileUtils { private static final Logger LOGGER = LoggerFactory.getLogger(FileUtils.class); @@ -111,13 +112,30 @@ public class FileUtils { } public static void deleteFileOrDirectory(File file, boolean quietForNoSuchFile) { + deleteFileOrDirectory(file, quietForNoSuchFile, null); + } + + public static void deleteFileOrDirectoryWithRateLimiter( + File file, LongConsumer deleteRateLimiter) { + deleteFileOrDirectory(file, false, deleteRateLimiter); + } + + public static void deleteFileOrDirectoryWithRateLimiter( + File file, boolean quietForNoSuchFile, LongConsumer deleteRateLimiter) { + deleteFileOrDirectory(file, quietForNoSuchFile, deleteRateLimiter); + } + + private static void deleteFileOrDirectory( + File file, boolean quietForNoSuchFile, LongConsumer deleteRateLimiter) { if (file.isDirectory()) { File[] files = file.listFiles(); if (files != null) { for (File subfile : files) { - deleteFileOrDirectory(subfile, quietForNoSuchFile); + deleteFileOrDirectory(subfile, quietForNoSuchFile, deleteRateLimiter); } } + } else if (deleteRateLimiter != null && file.isFile()) { + deleteRateLimiter.accept(file.length()); } try { Files.delete(file.toPath()); @@ -161,7 +179,16 @@ public class FileUtils { } public static void deleteDirectoryAndEmptyParent(File folder) { - deleteFileOrDirectory(folder); + deleteDirectoryAndEmptyParent(folder, null); + } + + public static void deleteDirectoryAndEmptyParentWithRateLimiter( + File folder, LongConsumer deleteRateLimiter) { + deleteDirectoryAndEmptyParent(folder, deleteRateLimiter); + } + + private static void deleteDirectoryAndEmptyParent(File folder, LongConsumer deleteRateLimiter) { + deleteFileOrDirectory(folder, false, deleteRateLimiter); final File parentFolder = folder.getParentFile(); File[] files = parentFolder.listFiles(); if (parentFolder.isDirectory() && (files == null || files.length == 0)) { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java similarity index 62% copy from iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java copy to iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java index bb6b95b45ad..0e3e48f75fc 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/snapshot/IoTConsensusRateLimiter.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/utils/RegionMigrationRateLimiter.java @@ -17,18 +17,15 @@ * under the License. */ -package org.apache.iotdb.consensus.iot.snapshot; +package org.apache.iotdb.commons.utils; import com.google.common.util.concurrent.RateLimiter; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -public class IoTConsensusRateLimiter { - private static final Logger logger = LoggerFactory.getLogger(IoTConsensusRateLimiter.class); +public class RegionMigrationRateLimiter { private final RateLimiter rateLimiter = RateLimiter.create(Double.MAX_VALUE); - private IoTConsensusRateLimiter() {} + private RegionMigrationRateLimiter() {} public void init(long regionMigrationSpeedLimitBytesPerSecond) { rateLimiter.setRate( @@ -37,26 +34,21 @@ public class IoTConsensusRateLimiter { : regionMigrationSpeedLimitBytesPerSecond); } - /** - * Acquire the size of the data to be sent. - * - * @param transitDataSize the size of the data to be sent - */ - public void acquireTransitDataSizeWithRateLimiter(long transitDataSize) { - while (transitDataSize > 0) { - if (transitDataSize > Integer.MAX_VALUE) { + public void acquire(long sizeInBytes) { + while (sizeInBytes > 0) { + if (sizeInBytes > Integer.MAX_VALUE) { rateLimiter.acquire(Integer.MAX_VALUE); - transitDataSize -= Integer.MAX_VALUE; + sizeInBytes -= Integer.MAX_VALUE; } else { - rateLimiter.acquire((int) transitDataSize); + rateLimiter.acquire((int) sizeInBytes); return; } } } - private static final IoTConsensusRateLimiter INSTANCE = new IoTConsensusRateLimiter(); + private static final RegionMigrationRateLimiter INSTANCE = new RegionMigrationRateLimiter(); - public static IoTConsensusRateLimiter getInstance() { + public static RegionMigrationRateLimiter getInstance() { return INSTANCE; } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java index 4c2210998b0..f2df776d02a 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/utils/FileUtilsTest.java @@ -33,6 +33,8 @@ import java.io.File; import java.io.IOException; import java.nio.file.Files; import java.util.Collections; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; public class FileUtilsTest { private File tmpDir; @@ -72,6 +74,45 @@ public class FileUtilsTest { Assert.assertNull(FileUtils.getIllegalError4Directory("valid_dir")); } + @Test + public void testDeleteFileOrDirectoryWithRateLimiter() throws IOException { + File deleteDir = new File(tmpDir, "deleteWithRateLimiter"); + File subDir = new File(deleteDir, "subDir"); + Assert.assertTrue(subDir.mkdirs()); + Files.write(new File(deleteDir, "file1").toPath(), new byte[3]); + Files.write(new File(subDir, "file2").toPath(), new byte[7]); + Files.write(new File(subDir, "empty").toPath(), new byte[0]); + + AtomicLong acquiredBytes = new AtomicLong(); + AtomicInteger acquiredFiles = new AtomicInteger(); + FileUtils.deleteFileOrDirectoryWithRateLimiter( + deleteDir, + 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(
