Skip to content

Make IoTConsensus peer-list reset retries best effort and fix Daily CI regressions #18929

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
CRZbulabula merged 1 commit into from
Oct 20, 2026
Merged

Make IoTConsensus peer-list reset retries best effort and fix Daily CI regressions #18929

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions ...-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.";
Expand Down
2 changes: 0 additions & 2 deletions ...-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 次后失败。";
Expand Down
45 changes: 23 additions & 22 deletions ...ain/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand All @@ -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) {
Expand Down
72 changes: 47 additions & 25 deletions ...st/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)));
Expand Down Expand Up @@ -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(
Expand Down
10 changes: 8 additions & 2 deletions ...ava/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down
52 changes: 46 additions & 6 deletions ...t/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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));
Expand Down Expand Up @@ -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)
Expand All @@ -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);
}
Expand All @@ -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();
Expand All @@ -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));
}

Expand Down
Loading