use leader as coordinator & fix IT

apache · Jan 17, 2025 · ea630f1 · ea630f1
1 parent c4b4fc9
commit ea630f1
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 6 deletions.
diff --git a/...che/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java b/...che/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionReconstructForIoTV1IT.java
@@ -103,12 +103,13 @@ public void normal1C3DTest() throws Exception {
       session.open();
 
       // delete one DataNode's data dir, stop another DataNode
-      FileUtils.deleteDirectory(
+      File dataDirToBeReconstructed =
           new File(
               EnvFactory.getEnv()
                   .dataNodeIdToWrapper(dataNodeToBeReconstructed)
                   .get()
-                  .getDataPath()));
+                  .getDataPath());
+      FileUtils.deleteDirectory(dataDirToBeReconstructed);
       EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeToBeClosed).get().stopForcibly();
 
       // now, the query should throw exception
@@ -121,11 +122,13 @@ public void normal1C3DTest() throws Exception {
       EnvFactory.getAbstractEnv().checkNodeInStatus(dataNodeToBeClosed, NodeStatus.Running);
       session.executeNonQueryStatement(
           String.format(RECONSTRUCT_FORMAT, selectedRegion, dataNodeToBeReconstructed));
-      Thread.sleep(5000);
       Awaitility.await()
           .pollInterval(1, TimeUnit.SECONDS)
           .atMost(1, TimeUnit.MINUTES)
-          .until(() -> getRegionStatusWithoutRunning(session).isEmpty());
+          .until(
+              () ->
+                  getRegionStatusWithoutRunning(session).isEmpty()
+                      && dataDirToBeReconstructed.exists());
       EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeToBeClosed).get().stopForcibly();
 
       // now, the query should work fine

diff --git a/...ignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/...ignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java
@@ -577,12 +577,19 @@ public Optional<TDataNodeLocation> filterDataNodeWithOtherRegionReplica(
         configManager.getNodeManager().filterDataNodeThroughStatus(allowingStatus).stream()
             .map(TDataNodeConfiguration::getLocation)
             .collect(Collectors.toList());
+    final int leaderId = configManager.getLoadManager().getRegionLeaderMap().get(regionId);
     Collections.shuffle(aliveDataNodes);
+    Optional<TDataNodeLocation> bestChoice = Optional.empty();
     for (TDataNodeLocation aliveDataNode : aliveDataNodes) {
       if (regionLocations.contains(aliveDataNode) && !excludeLocations.contains(aliveDataNode)) {
-        return Optional.of(aliveDataNode);
+        if (leaderId == aliveDataNode.getDataNodeId()) {
+          bestChoice = Optional.of(aliveDataNode);
+          break;
+        } else if (!bestChoice.isPresent()) {
+          bestChoice = Optional.of(aliveDataNode);
+        }
       }
     }
-    return Optional.empty();
+    return bestChoice;
   }
 }