From 415c918b0f9c9b5e310565bfc1504f24ccb92ab5 Mon Sep 17 00:00:00 2001 From: Di Wang Date: Tue, 20 Oct 2026 12:18:16 +0800 Subject: [PATCH] Make IoTConsensus peer-list reset retries best effort 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. --- .../confignode/i18n/ProcedureMessages.java | 2 - .../confignode/i18n/ProcedureMessages.java | 2 - .../region/RemoveRegionPeerProcedure.java | 45 ++++++------ .../node/RemoveDataNodesProcedureTest.java | 72 ++++++++++++------- .../RemovePeerRestartConsensusTest.java | 10 ++- .../region/RestartDuringRemovePeerTest.java | 52 ++++++++++++-- 6 files changed, 124 insertions(+), 59 deletions(-) diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 3874fbaac0d11..737c1c49bb4bc 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/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."; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 394480f654325..fe56909718971 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/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 次后失败。"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java index 01993f914408f..ffea2aa044c3a 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java +++ b/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,7 +218,8 @@ 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: @@ -216,8 +227,11 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionPeerStat } } 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; } @@ -225,28 +239,15 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionPeerStat 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) { diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java index 2769701c0d4b3..916270abf2217 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java +++ b/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,36 +306,49 @@ 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. @@ -339,16 +356,21 @@ private void assertFinalizationFailureCleanup( "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( diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java index b03f06c95a2c6..64f1eeecf6347 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java +++ b/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() diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java index be868283edcc8..1ae8050bc01e0 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java +++ b/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)); }