IoTConsensus hashes the group ID to assign to fixed directories.

Author: hongzhi-gao

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensus.java

@@ -76,6 +76,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -173,29 +174,76 @@ private void initAndRecover() throws IOException {
         if (!storageDir.mkdirs()) {
           throw new IOException(String.format("Unable to create consensus dir at %s", storageDir));
         }
-      } else {
+      }
+    }
+
+    Map<ConsensusGroupId, String> consensusGroupDirs = new HashMap<>();
+
+    for (String baseFolder : folderManager.getFolders()) {
+      File storageDir = new File(baseFolder);
+      if (storageDir.exists()) {
         try (DirectoryStream<Path> stream = Files.newDirectoryStream(storageDir.toPath())) {
           for (Path path : stream) {
+            if (!Files.isDirectory(path)) {
+              continue;
+            }
+
             String[] items = path.getFileName().toString().split("_");
-            ConsensusGroupId consensusGroupId =
-                ConsensusGroupId.Factory.create(
-                    Integer.parseInt(items[0]), Integer.parseInt(items[1]));
-            IoTConsensusServerImpl consensus =
-                new IoTConsensusServerImpl(
-                    path.toString(),
-                    new Peer(consensusGroupId, thisNodeId, thisNode),
-                    new TreeSet<>(),
-                    registry.apply(consensusGroupId),
-                    backgroundTaskService,
-                    clientManager,
-                    syncClientManager,
-                    config);
-            stateMachineMap.put(consensusGroupId, consensus);
+            if (items.length != 2) {
+              continue; // Skip directories that do not meet the naming convention.
+            }
+
+            try {
+              ConsensusGroupId consensusGroupId =
+                  ConsensusGroupId.Factory.create(
+                      Integer.parseInt(items[0]), Integer.parseInt(items[1]));
+              String expectedFolder;
+              try {
+                expectedFolder = folderManager.getFolderByHashId(consensusGroupId.getId());
+              } catch (DiskSpaceInsufficientException e) {
+                logger.warn(
+                    "Cannot determine expected folder for group {}, skipping", consensusGroupId);
+                continue;
+              }
+
+              // Use this directory if it is in the correct baseFolder where the consensus group
+              // should be located.
+              if (baseFolder.equals(expectedFolder)) {
+                consensusGroupDirs.put(consensusGroupId, path.toString());
+              } else {
+                logger.warn(
+                    "Found consensus group {} in wrong folder: {} (should be in {}), skipping",
+                    consensusGroupId,
+                    baseFolder,
+                    expectedFolder);
+              }
+            } catch (NumberFormatException e) {
+              logger.warn("Invalid consensus group directory name: {}", path.getFileName());
+            }
           }
+        } catch (IOException e) {
+          logger.error("Error scanning directory: {}", baseFolder, e);
         }
       }
     }
 
+    for (Map.Entry<ConsensusGroupId, String> entry : consensusGroupDirs.entrySet()) {
+      ConsensusGroupId consensusGroupId = entry.getKey();
+      String path = entry.getValue();
+
+      IoTConsensusServerImpl consensus =
+          new IoTConsensusServerImpl(
+              path,
+              new Peer(consensusGroupId, thisNodeId, thisNode),
+              new TreeSet<>(),
+              registry.apply(consensusGroupId),
+              backgroundTaskService,
+              clientManager,
+              syncClientManager,
+              config);
+      stateMachineMap.put(consensusGroupId, consensus);
+    }
+
     if (correctPeerListBeforeStart != null) {
       BiConsumer<ConsensusGroupId, List<Peer>> resetPeerListWithoutThrow =
           (consensusGroupId, peers) -> {
@@ -284,7 +332,7 @@ public void createLocalPeer(ConsensusGroupId groupId, List<Peer> peers)
 
                   String path = null;
                   try {
-                    path = buildPeerDir(folderManager.getNextFolder(), groupId);
+                    path = buildPeerDir(folderManager.getFolderByHashId(groupId.getId()), groupId);
                   } catch (DiskSpaceInsufficientException e) {
                     logger.warn(
                         "Failed to create consensus directory for group {} due to disk space insufficiency: {}",
@@ -293,6 +341,7 @@ public void createLocalPeer(ConsensusGroupId groupId, List<Peer> peers)
                     return null;
                   }
                   File file = new File(path);
+                  // debug print the path of consensus dir
                   System.out.println(file.getAbsolutePath());
                   if (!file.mkdirs()) {
                     logger.warn("Unable to create consensus dir for group {} at {}", groupId, path);
@@ -336,8 +385,13 @@ public void deleteLocalPeer(ConsensusGroupId groupId) throws ConsensusException
     if (!exist.get()) {
       throw new ConsensusGroupNotExistException(groupId);
     }
-    for (String folder : folderManager.getFolders()) {
-      FileUtils.deleteFileOrDirectory(new File(buildPeerDir(folder, groupId)));
+    try {
+      FileUtils.deleteFileOrDirectory(
+          new File(buildPeerDir(folderManager.getFolderByHashId(groupId.getId()), groupId)));
+    } catch (DiskSpaceInsufficientException e) {
+      logger.warn(
+          "Failed to delete consensus directory for group {} due to : {}", groupId, e.getMessage());
+      throw new ConsensusException(e);
     }
     KillPoint.setKillPoint(IoTConsensusDeleteLocalPeerKillPoints.AFTER_DELETE);
   }

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StabilityTest.java

@@ -262,8 +262,10 @@ public void multiRegionSnapshotTest() throws ConsensusException, DiskSpaceInsuff
     }
 
     for (int i = 0; i < folders.length; i++) {
-      File dataDir = new File(IoTConsensus.buildPeerDir(folders[i], dataRegionIds[i]));
-      System.out.println(dataDir.getAbsolutePath());
+      File dataDir =
+          new File(
+              IoTConsensus.buildPeerDir(
+                  folders[dataRegionIds[i].getId() % folders.length], dataRegionIds[i]));
       File[] versionFiles1 =
           dataDir.listFiles(
               (dir, name) -> name.startsWith(IoTConsensusServerImpl.SNAPSHOT_DIR_NAME));

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/disk/FolderManager.java

@@ -35,6 +35,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 public class FolderManager {
   private static final Logger logger = LoggerFactory.getLogger(FolderManager.class);
@@ -104,6 +105,37 @@ public String getNextFolder() throws DiskSpaceInsufficientException {
     }
   }
 
+  /**
+   * Get folder by hash ID (typically consensus group ID), distributing as evenly as possible among
+   * healthy folders using hash-based distribution
+   *
+   * @param hashId the hash ID (typically consensus group ID) used to determine folder selection
+   * @return the selected folder path
+   * @throws DiskSpaceInsufficientException if no healthy folders are available
+   */
+  public String getFolderByHashId(int hashId) throws DiskSpaceInsufficientException {
+    if (folders.isEmpty()) {
+      throw new DiskSpaceInsufficientException(folders);
+    }
+
+    List<String> healthyFolders =
+        folders.stream()
+            .filter(
+                folder ->
+                    foldersStates.getOrDefault(folder, FolderState.ABNORMAL) == FolderState.HEALTHY)
+            .collect(Collectors.toList());
+
+    if (healthyFolders.isEmpty()) {
+      logger.error("No healthy folders available, change system mode to read-only.");
+      CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly);
+      CommonDescriptor.getInstance().getConfig().setStatusReason(NodeStatus.DISK_FULL);
+      throw new DiskSpaceInsufficientException(folders);
+    }
+
+    int index = Math.abs(hashId) % healthyFolders.size();
+    return healthyFolders.get(index);
+  }
+
   boolean hasHealthyFolder() {
     return folders.stream()
         .anyMatch(