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/2] 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/2] 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