Make IoTConsensus peer-list reset retries best effort (#18929)

Finish node removal after bounded peer-list reset retries are exhausted.
Keep metadata failures fatal and simplify caller-controlled retry handling.
Cover exhaustion paths and adapt regression fixtures to targeted region lookups.

Author: d-wang-commit

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -853,8 +853,6 @@ public final class ProcedureMessages {
   public static final String
       LOG_FAILED_TO_RESET_PEER_LIST_FOR_ARG_ON_DATANODE_ARG_TO_ARG_ARG_C8D872B4 =
           "Failed to reset peer list for {} on DataNode {} to {}: {}";
-  public static final String LOG_PIDARG_REMOVEREGION_STATE_ARG_FAILED_RETRY_ARG_ARG_IN_ARG_MS_41B8ECB5 =
-      "[pid{}][RemoveRegion] state {} failed, retry {}/{} in {} ms.";
   public static final String
       EXCEPTION_FAILED_TO_FINALIZE_REGION_PEER_REMOVAL_FOR_ARG_AT_STATE_ARG_AFTER_ARG_ATTEMPTS_382EA168 =
           "Failed to finalize Region peer removal for %s at state %s after %d attempts.";

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -812,8 +812,6 @@ public final class ProcedureMessages {
   public static final String
       LOG_FAILED_TO_RESET_PEER_LIST_FOR_ARG_ON_DATANODE_ARG_TO_ARG_ARG_C8D872B4 =
           "将 {} 在 DataNode {} 上的 peer 列表重置为 {} 失败：{}";
-  public static final String LOG_PIDARG_REMOVEREGION_STATE_ARG_FAILED_RETRY_ARG_ARG_IN_ARG_MS_41B8ECB5 =
-      "[pid{}][RemoveRegion] 状态 {} 失败，第 {}/{} 次重试将在 {} 毫秒后进行。";
   public static final String
       EXCEPTION_FAILED_TO_FINALIZE_REGION_PEER_REMOVAL_FOR_ARG_AT_STATE_ARG_AFTER_ARG_ATTEMPTS_382EA168 =
           "Region %s 的 peer 移除收尾在状态 %s 尝试 %d 次后失败。";

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java

@@ -39,6 +39,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.annotation.Nullable;
+
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -198,7 +200,15 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionPeerStat
         case REMOVE_REGION_LOCATION_CACHE:
           tsStatus = handler.removeRegionLocation(regionId, targetDataNode);
           if (tsStatus == null || tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
-            return retryStep(state);
+            return retryStep(
+                state,
+                new ProcedureException(
+                    String.format(
+                        ProcedureMessages
+                            .EXCEPTION_FAILED_TO_FINALIZE_REGION_PEER_REMOVAL_FOR_ARG_AT_STATE_ARG_AFTER_ARG_ATTEMPTS_382EA168,
+                        regionId,
+                        state,
+                        getCycles() + 1)));
           }
           if (!handler.isIoTConsensusV1DataRegion(regionId)) {
             return finishRemoval(state);
@@ -208,45 +218,36 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionPeerStat
           break;
         case RESET_IOT_CONSENSUS_V1_PEER_LIST:
           if (!handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, targetDataNode)) {
-            return retryStep(state);
+            // The replica metadata is already committed. Peer list reconciliation is best effort.
+            return retryStep(state, null);
           }
           return finishRemoval(state);
         default:
           throw new ProcedureException(ProcedureMessages.UNSUPPORTED_STATE + state.name());
       }
     } catch (Exception e) {
       LOGGER.error(ProcedureMessages.REMOVEREGIONPEER_STATE_FAILED, state, e);
-      if (state == REMOVE_REGION_LOCATION_CACHE || state == RESET_IOT_CONSENSUS_V1_PEER_LIST) {
-        return retryStep(state);
+      if (state == REMOVE_REGION_LOCATION_CACHE) {
+        return retryStep(state, new ProcedureException(e));
+      }
+      if (state == RESET_IOT_CONSENSUS_V1_PEER_LIST) {
+        return retryStep(state, null);
       }
       return Flow.NO_MORE_STATE;
     }
     LOGGER.info(ProcedureMessages.PID_REMOVEREGION_STATE_SUCCESS, getProcId(), state);
     return Flow.HAS_MORE_STATE;
   }
 
-  private Flow retryStep(RemoveRegionPeerState state) {
+  private Flow retryStep(
+      RemoveRegionPeerState state, @Nullable ProcedureException failureOnExhaustion) {
     if (getCycles() >= MAX_FINALIZATION_RETRY) {
-      ProcedureException failure =
-          new ProcedureException(
-              String.format(
-                  ProcedureMessages
-                      .EXCEPTION_FAILED_TO_FINALIZE_REGION_PEER_REMOVAL_FOR_ARG_AT_STATE_ARG_AFTER_ARG_ATTEMPTS_382EA168,
-                  regionId,
-                  state,
-                  getCycles() + 1));
-      LOGGER.error(ProcedureMessages.REMOVEREGIONPEER_STATE_FAILED, state, failure);
-      setFailure(failure);
+      if (failureOnExhaustion != null) {
+        setFailure(failureOnExhaustion);
+      }
       return Flow.NO_MORE_STATE;
     }
     setNextState(state);
-    LOGGER.warn(
-        ProcedureMessages.LOG_PIDARG_REMOVEREGION_STATE_ARG_FAILED_RETRY_ARG_ARG_IN_ARG_MS_41B8ECB5,
-        getProcId(),
-        state,
-        getCycles() + 1,
-        MAX_FINALIZATION_RETRY,
-        FINALIZATION_RETRY_INTERVAL_MS);
     try {
       Thread.sleep(FINALIZATION_RETRY_INTERVAL_MS);
     } catch (InterruptedException e) {

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java

@@ -137,17 +137,19 @@ public void serDeTest() throws IOException {
   }
 
   @Test
-  public void metadataRetryExhaustionCleansUpNodeRemoval() throws Exception {
-    assertFinalizationFailureCleanup(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE, false);
+  public void metadataRetryExhaustionRetriesFailedNodeStatusCleanup() throws Exception {
+    assertFinalizationExhaustionCleanup(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE, true);
   }
 
   @Test
-  public void peerListRetryExhaustionRetriesFailedNodeStatusCleanup() throws Exception {
-    assertFinalizationFailureCleanup(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, true);
+  public void peerListRetryExhaustionCompletesNodeRemoval() throws Exception {
+    assertFinalizationExhaustionCleanup(
+        RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, false);
   }
 
-  private void assertFinalizationFailureCleanup(
+  private void assertFinalizationExhaustionCleanup(
       RemoveRegionPeerState failingState, boolean failFirstStatusRestoration) throws Exception {
+    boolean metadataFailure = failingState == RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE;
     TDataNodeLocation runningNode = fromDataNodeLocation;
     TDataNodeLocation readOnlyNode = location(12);
     TDataNodeLocation drainedNode = location(13);
@@ -178,10 +180,12 @@ private void assertFinalizationFailureCleanup(
         "root.remove",
         new TRegionReplicaSet(
             consensusGroupId, new ArrayList<>(Arrays.asList(runningNode, retainedNode))));
-    groups.addRegionGroup(
-        "root.remove",
-        new TRegionReplicaSet(
-            pendingRunningRegion, new ArrayList<>(Arrays.asList(runningNode, retainedNode))));
+    if (metadataFailure) {
+      groups.addRegionGroup(
+          "root.remove",
+          new TRegionReplicaSet(
+              pendingRunningRegion, new ArrayList<>(Arrays.asList(runningNode, retainedNode))));
+    }
     groups.addRegionGroup(
         "root.remove",
         new TRegionReplicaSet(
@@ -229,7 +233,7 @@ private void assertFinalizationFailureCleanup(
     when(regionHandler.removeRegionLocation(consensusGroupId, runningNode))
         .thenAnswer(
             ignored ->
-                failingState == RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE
+                metadataFailure
                     ? new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode())
                     : partitionInfo.removeRegionLocation(
                         new RemoveRegionLocationPlan(consensusGroupId, runningNode)));
@@ -302,53 +306,71 @@ private void assertFinalizationFailureCleanup(
           new RemoveDataNodesProcedure(removalTargets, originalStatuses);
       long rootId = executor.submitProcedure(root);
       ProcedureTestUtil.waitForProcedure(executor, rootId);
-      Assert.assertTrue("Node removal must finish failure cleanup", executor.isFinished(rootId));
-      Assert.assertEquals(ProcedureState.ROLLEDBACK, root.getState());
-      Assert.assertFalse(root.isSuccess());
-      RemoveRegionPeerProcedure failedChild =
+      Assert.assertTrue("Node removal must finish cleanup", executor.isFinished(rootId));
+      RemoveRegionPeerProcedure removalChild =
           store.recorded.values().stream()
               .filter(RemoveRegionPeerProcedure.class::isInstance)
               .map(RemoveRegionPeerProcedure.class::cast)
               .findFirst()
               .get();
-      Procedure<?> migrationParent = store.recorded.get(failedChild.getParentProcId());
+      Procedure<?> migrationParent = store.recorded.get(removalChild.getParentProcId());
       Assert.assertTrue(migrationParent instanceof RegionMigrateProcedure);
       Assert.assertEquals(rootId, migrationParent.getParentProcId());
-      Assert.assertSame(failedChild.getException(), root.getException());
-      Assert.assertTrue(root.getException().getMessage().contains(failingState.name()));
-      if (failingState == RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE) {
+      if (metadataFailure) {
+        Assert.assertEquals(ProcedureState.ROLLEDBACK, root.getState());
+        Assert.assertFalse(root.isSuccess());
+        Assert.assertSame(removalChild.getException(), root.getException());
+        Assert.assertTrue(root.getException().getMessage().contains(failingState.name()));
         verify(regionHandler, times(4)).removeRegionLocation(consensusGroupId, runningNode);
         verify(regionHandler, never()).resetPeerListAfterRemoveForIoTConsensusV1(any(), any());
       } else {
+        Assert.assertTrue(root.isSuccess());
+        Assert.assertTrue(migrationParent.isSuccess());
+        Assert.assertTrue(removalChild.isSuccess());
+        Assert.assertNull(root.getException());
+        Assert.assertNull(migrationParent.getException());
+        Assert.assertNull(removalChild.getException());
         verify(regionHandler).removeRegionLocation(consensusGroupId, runningNode);
         verify(regionHandler, times(4))
             .resetPeerListAfterRemoveForIoTConsensusV1(consensusGroupId, runningNode);
       }
 
-      // Failed children bypass STOP_DATA_NODE: root rollback must still finalize node statuses.
-      Assert.assertEquals(NodeStatus.Running, currentStatuses.get(runningNode.getDataNodeId()));
+      // Metadata failures use rollback cleanup; exhausted peer-list resets use normal
+      // STOP_DATA_NODE.
+      if (metadataFailure) {
+        Assert.assertEquals(NodeStatus.Running, currentStatuses.get(runningNode.getDataNodeId()));
+      }
       Assert.assertEquals(NodeStatus.ReadOnly, currentStatuses.get(readOnlyNode.getDataNodeId()));
-      Assert.assertTrue(registeredNodes.contains(runningNode.getDataNodeId()));
+      Assert.assertEquals(metadataFailure, registeredNodes.contains(runningNode.getDataNodeId()));
       Assert.assertTrue(registeredNodes.contains(readOnlyNode.getDataNodeId()));
       Assert.assertFalse(registeredNodes.contains(drainedNode.getDataNodeId()));
-      Assert.assertEquals(Collections.singleton(drainedNode.getDataNodeId()), stoppedNodes);
+      Assert.assertEquals(
+          metadataFailure
+              ? Collections.singleton(drainedNode.getDataNodeId())
+              : Set.of(runningNode.getDataNodeId(), drainedNode.getDataNodeId()),
+          stoppedNodes);
       Assert.assertEquals(failFirstStatusRestoration ? 2 : 1, restorationAttempts.get());
       if (failFirstStatusRestoration) {
         // A node-lock wakeup must not bypass the five-second rollback retry timeout.
         Assert.assertTrue(
             "Status restoration must wait for the rollback retry timeout",
             restorationRetryDelay.get() >= TimeUnit.SECONDS.toNanos(4));
       }
-      verify(nodeHandler).broadcastDataNodeStatusChange(Arrays.asList(runningNode, readOnlyNode));
+      verify(nodeHandler)
+          .broadcastDataNodeStatusChange(
+              metadataFailure
+                  ? Arrays.asList(runningNode, readOnlyNode)
+                  : Collections.singletonList(readOnlyNode));
 
       // Cleanup must not undo the completed add peer or a committed removal.
       Assert.assertTrue(
           partitionManager.isDataNodeContainsRegion(
               toDataNodeLocation.getDataNodeId(), consensusGroupId));
       Assert.assertEquals(
-          failingState == RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE,
+          metadataFailure,
           partitionManager.isDataNodeContainsRegion(runningNode.getDataNodeId(), consensusGroupId));
-      Assert.assertTrue(
+      Assert.assertEquals(
+          metadataFailure,
           partitionManager.isDataNodeContainsRegion(
               runningNode.getDataNodeId(), pendingRunningRegion));
       Assert.assertTrue(

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java

@@ -159,8 +159,14 @@ public void removalProcedureRepairsRestartedPeerAndWritesContinue() throws Excep
       final PartitionInfo partitionInfo = newPartitionInfo(locations);
       final ConfigManager configManager =
           Mockito.mock(ConfigManager.class, Mockito.RETURNS_DEEP_STUBS);
-      Mockito.when(configManager.getPartitionManager().getAllReplicaSets())
-          .thenAnswer(ignored -> partitionInfo.getAllReplicaSets());
+      Mockito.when(configManager.getPartitionManager().getRegionDatabase(REGION))
+          .thenAnswer(call -> partitionInfo.getRegionDatabase(call.getArgument(0)));
+      Mockito.when(
+              configManager
+                  .getPartitionManager()
+                  .getReplicaSets("root.restart_repair", Collections.singletonList(REGION)))
+          .thenAnswer(
+              call -> partitionInfo.getReplicaSets(call.getArgument(0), call.getArgument(1)));
       Mockito.when(
               configManager
                   .getPartitionManager()

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java

@@ -66,6 +66,7 @@
 import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyList;
@@ -117,8 +118,10 @@ public void setUp() throws Exception {
     loadManager = Mockito.mock(LoadManager.class, Mockito.RETURNS_DEEP_STUBS);
     when(configManager.getPartitionManager()).thenReturn(partitionManager);
     when(configManager.getLoadManager()).thenReturn(loadManager);
-    when(partitionManager.getAllReplicaSets())
-        .thenAnswer(ignored -> partitionInfo.getAllReplicaSets());
+    when(partitionManager.getRegionDatabase(regionId))
+        .thenAnswer(call -> partitionInfo.getRegionDatabase(call.getArgument(0)));
+    when(partitionManager.getReplicaSets("root.restart", Collections.singletonList(regionId)))
+        .thenAnswer(call -> partitionInfo.getReplicaSets(call.getArgument(0), call.getArgument(1)));
     when(partitionManager.removeRegionLocation(any()))
         .thenAnswer(call -> partitionInfo.removeRegionLocation(call.getArgument(0)));
     handler = Mockito.spy(new RegionMaintainHandler(configManager));
@@ -286,7 +289,7 @@ public void replayAfterSuccessfulCorrectionIsIdempotent() throws Exception {
   }
 
   @Test
-  public void offlineReplicaFailsAfterThreeRetriesInsteadOfWaitingForever() throws Exception {
+  public void offlineReplicaCompletesRemovalAfterThreeRetries() throws Exception {
     AccessibleRemoveProcedure procedure = beforePeerListReset();
     doReturn(Collections.singletonMap(coordinator.getDataNodeId(), success()))
         .when(handler)
@@ -297,8 +300,27 @@ public void offlineReplicaFailsAfterThreeRetriesInsteadOfWaitingForever() throws
       assertFalse(procedure.isFailed());
     }
     assertEquals(0, procedure.doExecute(env).length);
-    assertTrue(procedure.isFailed());
-    assertTrue(procedure.getException().getMessage().contains("4"));
+    assertFalse(procedure.isFailed());
+    assertNull(procedure.getException());
+    assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations());
+    verify(handler, times(4)).resetPeerList(eq(regionId), eq(finalPeers), anyMap());
+    verify(handler).removeRegionLocation(regionId, removedNode);
+  }
+
+  @Test
+  public void repeatedCorrectionExceptionsAlsoCompleteRemovalAfterThreeRetries() throws Exception {
+    AccessibleRemoveProcedure procedure = beforePeerListReset();
+    doThrow(new IllegalStateException("injected RPC failure"))
+        .when(handler)
+        .resetPeerList(any(), anyList(), anyMap());
+    for (int retry = 0; retry < 3; retry++) {
+      assertEquals(1, procedure.doExecute(env).length);
+      assertFalse(procedure.isFailed());
+    }
+    assertEquals(0, procedure.doExecute(env).length);
+    assertFalse(procedure.isFailed());
+    assertNull(procedure.getException());
+    assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations());
     verify(handler, times(4)).resetPeerList(eq(regionId), eq(finalPeers), anyMap());
     verify(handler).removeRegionLocation(regionId, removedNode);
   }
@@ -317,6 +339,24 @@ public void metadataFailureAlsoStopsAtRetryLimitWithoutResettingPeers() throws E
     verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
   }
 
+  @Test
+  public void metadataExceptionsStillFailAtRetryLimitWithoutResettingPeers() throws Exception {
+    AccessibleRemoveProcedure procedure = beforeMetadataRemoval();
+    IllegalStateException failure = new IllegalStateException("injected metadata failure");
+    doThrow(failure).when(partitionManager).removeRegionLocation(any());
+    for (int retry = 0; retry < 3; retry++) {
+      assertEquals(1, procedure.doExecute(env).length);
+      assertFalse(procedure.isFailed());
+    }
+    assertEquals(0, procedure.doExecute(env).length);
+    assertTrue(procedure.isFailed());
+    assertEquals(failure, procedure.getException().getCause());
+    assertTrue(
+        partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations().contains(removedNode));
+    verify(handler, times(4)).removeRegionLocation(regionId, removedNode);
+    verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
+  }
+
   @Test
   public void interruptedRetryRetainsItsStepForRecovery() throws Exception {
     AccessibleRemoveProcedure procedure = beforePeerListReset();
@@ -342,7 +382,7 @@ public void correctionRejectsStaleMetadataAndNullResponsesButAcceptsDeletedRegio
     partitionInfo.removeRegionLocation(new RemoveRegionLocationPlan(regionId, removedNode));
     doReturn(null).when(handler).resetPeerList(any(), anyList(), anyMap());
     assertFalse(handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, removedNode));
-    when(partitionManager.getAllReplicaSets()).thenReturn(Collections.emptyList());
+    when(partitionManager.getRegionDatabase(regionId)).thenReturn(null);
     assertTrue(handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, removedNode));
   }