From e68b2f9741df77b98436cdb96591bbe98dc5dc90 Mon Sep 17 00:00:00 2001 From: Di Wang Date: Fri, 16 Oct 2026 18:37:05 +0800 Subject: [PATCH] Fix stale IoTConsensus peers after restart during removal Commit removal metadata before reconciling IoTConsensus V1 peer lists. Retry metadata updates and V1 resets up to three times before failing. Cover restart recovery and filter V1-only crash points in Ratis/V2 ITs. --- ...ionMigrateConfigNodeCrashIoTV2BatchIT.java | 2 + ...onMigrateConfigNodeCrashIoTV2StreamIT.java | 2 + ...egionMigrateConfigNodeCrashForRatisIT.java | 2 + .../confignode/i18n/ProcedureMessages.java | 11 + .../confignode/i18n/ProcedureMessages.java | 11 + .../procedure/env/RegionMaintainHandler.java | 55 +- .../region/RemoveRegionPeerProcedure.java | 77 ++- .../state/RemoveRegionPeerState.java | 2 + .../RemovePeerRestartConsensusTest.java | 634 ++++++++++++++++++ .../region/RestartDuringRemovePeerTest.java | 450 +++++++++++++ .../StalePeerListRestartReproductionTest.java | 272 ++++++++ 11 files changed, 1505 insertions(+), 13 deletions(-) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java create mode 100644 iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StalePeerListRestartReproductionTest.java diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java index 5b81272acb998..c484931e75048 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java @@ -173,6 +173,8 @@ public void cnCrashTest() throws Exception { .collect(Collectors.toList())); killConfigNodeKeywords.addAll( Arrays.stream(RemoveRegionPeerState.values()) + // IoTConsensus V2 finishes removal before the V1-only reset state. + .filter(state -> state != RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST) .map(KillPoint::enumToString) .collect(Collectors.toList())); successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints(), KillNode.CONFIG_NODE); diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java index fc87df6eb83a8..a826b528b2355 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java @@ -187,6 +187,8 @@ public void cnCrashTest() throws Exception { .collect(Collectors.toList())); killConfigNodeKeywords.addAll( Arrays.stream(RemoveRegionPeerState.values()) + // IoTConsensus V2 finishes removal before the V1-only reset state. + .filter(state -> state != RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST) .map(KillPoint::enumToString) .collect(Collectors.toList())); successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints(), KillNode.CONFIG_NODE); diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java index 975e1611109d5..304ad079dfecc 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java @@ -183,6 +183,8 @@ public void cnCrashTest() throws Exception { .collect(Collectors.toList())); killConfigNodeKeywords.addAll( Arrays.stream(RemoveRegionPeerState.values()) + // Ratis finishes removal before the IoTConsensus V1-only reset state. + .filter(state -> state != RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST) .map(KillPoint::enumToString) .collect(Collectors.toList())); successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints(), KillNode.CONFIG_NODE); 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 18aca01243908..3874fbaac0d11 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 @@ -847,6 +847,17 @@ public final class ProcedureMessages { "RemoveRegionLocation remove region {} from DataNode {}, result is {}"; public static final String REMOVEREGIONPEER_STATE_FAILED = "RemoveRegionPeer state {} failed"; public static final String REMOVEREGIONPEER_STATE_SUCCESS = "RemoveRegionPeer state {} success"; + public static final String + LOG_CANNOT_RECONCILE_PEER_LIST_FOR_ARG_AFTER_REMOVING_DATANODE_ARG_THE_CURRENT_REPLICA_SET_STILL_CONTAINS_THE_REMOVED_DATANODE_DBBF45DC = + "Cannot reconcile peer list for {} after removing DataNode {}: the current replica set still contains the removed DataNode."; + 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."; public static final String REMOVEREGION_RATIS_TRANSFER_LEADER_FAIL_BUT_PROCEDURE_WILL_CONTINUE = "[RemoveRegion] Ratis transfer leader fail, but procedure will continue."; public static final String REMOVE_CONFIG_NODE = "Remove Config Node"; 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 14e52d72b9efe..394480f654325 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 @@ -806,6 +806,17 @@ public final class ProcedureMessages { "RemoveRegionLocation：从 DataNode {} 移除 region {}，结果为 {}"; public static final String REMOVEREGIONPEER_STATE_FAILED = "RemoveRegionPeer 状态 {} 失败"; public static final String REMOVEREGIONPEER_STATE_SUCCESS = "RemoveRegionPeer 状态 {} 成功"; + public static final String + LOG_CANNOT_RECONCILE_PEER_LIST_FOR_ARG_AFTER_REMOVING_DATANODE_ARG_THE_CURRENT_REPLICA_SET_STILL_CONTAINS_THE_REMOVED_DATANODE_DBBF45DC = + "无法纠正 {} 在移除 DataNode {} 后的 peer 列表：当前副本集仍包含已移除的 DataNode。"; + 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 次后失败。"; public static final String REMOVEREGION_RATIS_TRANSFER_LEADER_FAIL_BUT_PROCEDURE_WILL_CONTINUE = "[RemoveRegion] Ratis 转换 leader 失败，但 procedure 将继续。"; public static final String REMOVE_CONFIG_NODE = "移除 ConfigNode"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java index f378ef132182d..12791f260edb8 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java @@ -360,6 +360,58 @@ public Map resetPeerList( return clientHandler.getResponseMap(); } + public boolean isIoTConsensusV1DataRegion(TConsensusGroupId regionId) { + return TConsensusGroupType.DataRegion.equals(regionId.getType()) + && IOT_CONSENSUS.equals(CONF.getDataRegionConsensusProtocolClass()); + } + + /** + * Reconcile retained IoTConsensus V1 replicas after the removed location has been committed. A + * restarting replica may have received the previous peer list while the removal was in progress. + * The existing reset RPC waits for consensus startup before applying this final list. + */ + public boolean resetPeerListAfterRemoveForIoTConsensusV1( + TConsensusGroupId regionId, TDataNodeLocation removedDataNode) { + if (!isIoTConsensusV1DataRegion(regionId)) { + return true; + } + Optional replicaSet = getRegionReplicaSet(regionId); + if (!replicaSet.isPresent()) { + // Database deletion may have removed the entire Region from ConfigNode metadata. Its + // deletion tasks own the cleanup, and there are no retained replicas left to reconcile. + return true; + } + if (replicaSet.get().getDataNodeLocations().stream() + .anyMatch(location -> location.getDataNodeId() == removedDataNode.getDataNodeId())) { + LOGGER.warn( + ProcedureMessages + .LOG_CANNOT_RECONCILE_PEER_LIST_FOR_ARG_AFTER_REMOVING_DATANODE_ARG_THE_CURRENT_REPLICA_SET_STILL_CONTAINS_THE_REMOVED_DATANODE_DBBF45DC, + regionId, + removedDataNode.getDataNodeId()); + return false; + } + List correctLocations = replicaSet.get().getDataNodeLocations(); + Map retainedNodes = new HashMap<>(); + correctLocations.forEach(location -> retainedNodes.put(location.getDataNodeId(), location)); + // Include offline replicas so the caller can retry unsuccessful resets within its retry limit. + Map responses = resetPeerList(regionId, correctLocations, retainedNodes); + boolean allSucceeded = true; + for (TDataNodeLocation location : correctLocations) { + TSStatus response = responses == null ? null : responses.get(location.getDataNodeId()); + if (response == null || response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + LOGGER.warn( + ProcedureMessages + .LOG_FAILED_TO_RESET_PEER_LIST_FOR_ARG_ON_DATANODE_ARG_TO_ARG_ARG_C8D872B4, + regionId, + location.getDataNodeId(), + correctLocations, + response); + allSucceeded = false; + } + } + return allSucceeded; + } + // TODO: will use 'procedure yield' to refactor later public TRegionMigrateResult waitTaskFinish(long taskId, TDataNodeLocation dataNodeLocation) { return waitTaskFinish(taskId, dataNodeLocation, null); @@ -449,7 +501,7 @@ public void forceUpdateRegionCache( .forceUpdateRegionCache(regionId, newLocation.getDataNodeId(), regionStatus); } - public void removeRegionLocation( + public TSStatus removeRegionLocation( TConsensusGroupId regionId, TDataNodeLocation deprecatedLocation) { RemoveRegionLocationPlan req = new RemoveRegionLocationPlan(regionId, deprecatedLocation); TSStatus status = configManager.getPartitionManager().removeRegionLocation(req); @@ -460,6 +512,7 @@ public void removeRegionLocation( status); configManager.getLoadManager().removeRegionCache(regionId, deprecatedLocation.getDataNodeId()); configManager.getLoadManager().getRouteBalancer().balanceRegionLeaderAndPriority(); + return status; } /** 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 390eb63534c15..01993f914408f 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 @@ -49,12 +49,15 @@ import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.DROP_CONSENSUS_PIPES; import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE; import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.REMOVE_REGION_PEER; +import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST; import static org.apache.iotdb.rpc.TSStatusCode.SUCCESS_STATUS; public class RemoveRegionPeerProcedure extends RegionOperationProcedure { private static final Logger LOGGER = LoggerFactory.getLogger(RemoveRegionPeerProcedure.class); private static final int MAX_DELETE_OLD_REGION_PEER_RETRY = 3; private static final long DELETE_OLD_REGION_PEER_RETRY_INTERVAL_MS = 5_000; + private static final int MAX_FINALIZATION_RETRY = 3; + private static final long FINALIZATION_RETRY_INTERVAL_MS = 5_000; private TDataNodeLocation coordinator; private TDataNodeLocation targetDataNode; private transient int deleteOldRegionPeerAttempted = 0; @@ -193,30 +196,80 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionPeerStat setNextState(REMOVE_REGION_LOCATION_CACHE); break; case REMOVE_REGION_LOCATION_CACHE: - handler.removeRegionLocation(regionId, targetDataNode); + tsStatus = handler.removeRegionLocation(regionId, targetDataNode); + if (tsStatus == null || tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) { + return retryStep(state); + } + if (!handler.isIoTConsensusV1DataRegion(regionId)) { + return finishRemoval(state); + } setKillPoint(state); - LOGGER.info(ProcedureMessages.REMOVEREGIONPEER_STATE_SUCCESS, state); - LOGGER.info( - ProcedureMessages - .PID_REMOVEREGION_SUCCESS_REGION_HAS_BEEN_REMOVED_FROM_DATANODE_PROCEDURE, - getProcId(), - regionId.getId(), - targetDataNode.getDataNodeId(), - CommonDateTimeUtils.convertMillisecondToDurationStr( - System.currentTimeMillis() - getSubmittedTime()), - DateTimeUtils.convertLongToDate(getSubmittedTime(), "ms")); - return Flow.NO_MORE_STATE; + setNextState(RESET_IOT_CONSENSUS_V1_PEER_LIST); + break; + case RESET_IOT_CONSENSUS_V1_PEER_LIST: + if (!handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, targetDataNode)) { + return retryStep(state); + } + 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); + } return Flow.NO_MORE_STATE; } LOGGER.info(ProcedureMessages.PID_REMOVEREGION_STATE_SUCCESS, getProcId(), state); return Flow.HAS_MORE_STATE; } + private Flow retryStep(RemoveRegionPeerState state) { + 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); + 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) { + // Preserve the step for recovery when this ConfigNode stops or loses leadership. + Thread.currentThread().interrupt(); + } + return Flow.HAS_MORE_STATE; + } + + private Flow finishRemoval(RemoveRegionPeerState state) { + setKillPoint(state); + LOGGER.info(ProcedureMessages.REMOVEREGIONPEER_STATE_SUCCESS, state); + LOGGER.info( + ProcedureMessages.PID_REMOVEREGION_SUCCESS_REGION_HAS_BEEN_REMOVED_FROM_DATANODE_PROCEDURE, + getProcId(), + regionId.getId(), + targetDataNode.getDataNodeId(), + CommonDateTimeUtils.convertMillisecondToDurationStr( + System.currentTimeMillis() - getSubmittedTime()), + DateTimeUtils.convertLongToDate(getSubmittedTime(), "ms")); + return Flow.NO_MORE_STATE; + } + @Override protected void rollbackState(ConfigNodeProcedureEnv env, RemoveRegionPeerState state) throws IOException, InterruptedException, ProcedureException {} diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java index a9c463cbafe32..35e7cc956ecb0 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java @@ -25,4 +25,6 @@ public enum RemoveRegionPeerState { DELETE_OLD_REGION_PEER, REMOVE_REGION_LOCATION_CACHE, DROP_CONSENSUS_PIPES, + // Reconcile retained DataRegion replicas after removal, only for IoTConsensus V1. + RESET_IOT_CONSENSUS_V1_PEER_LIST, } 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 new file mode 100644 index 0000000000000..b03f06c95a2c6 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java @@ -0,0 +1,634 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure.impl.region; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TRegionMaintainTaskStatus; +import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.commons.request.IConsensusRequest; +import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager; +import org.apache.iotdb.confignode.conf.ConfigNodeConfig; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType; +import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan; +import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan; +import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.persistence.partition.PartitionInfo; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler; +import org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.consensus.IStateMachine; +import org.apache.iotdb.consensus.common.DataSet; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.common.request.BatchIndexedConsensusRequest; +import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest; +import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; +import org.apache.iotdb.consensus.config.ConsensusConfig; +import org.apache.iotdb.consensus.config.IoTConsensusConfig; +import org.apache.iotdb.consensus.iot.IoTConsensus; +import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; +import org.apache.iotdb.consensus.iot.log.ConsensusReqReader; +import org.apache.iotdb.consensus.iot.log.GetConsensusReqReaderPlan; +import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher; +import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher.LogDispatcherThread; +import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService; +import org.apache.iotdb.mpp.rpc.thrift.TRegionMigrateResult; +import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.thrift.protocol.TBinaryProtocol; +import org.apache.thrift.protocol.TCompactProtocol; +import org.apache.thrift.server.TThreadPoolServer; +import org.apache.thrift.transport.TServerSocket; +import org.apache.thrift.transport.layered.TFramedTransport; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.Mockito; + +import java.io.File; +import java.lang.reflect.Field; +import java.net.ServerSocket; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; + +/** + * Runs the ConfigNode removal state machine against real IoTConsensus replication services. Only + * ConfigNode persistence transport and the DataNode internal-service facade are adapted: metadata + * changes use PartitionInfo, and the production reset RPC crosses a loopback Thrift connection. The + * in-memory state machine stands in for DataRegion/WAL recovery; no full DataNode JVM is started. + */ +public class RemovePeerRestartConsensusTest { + + private static final DataRegionId GROUP = new DataRegionId(92); + private static final TConsensusGroupId REGION = GROUP.convertToTConsensusGroupId(); + private static final int INITIAL_WRITES = 8; + private static final int WRITES_DURING_STALE_MEMBERSHIP = 8; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void removalProcedureRepairsRestartedPeerAndWritesContinue() throws Exception { + final ConfigNodeConfig cnConfig = ConfigNodeDescriptor.getInstance().getConf(); + final String originalProtocol = cnConfig.getDataRegionConsensusProtocolClass(); + cnConfig.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS); + final ConcurrentMap servers = new ConcurrentHashMap<>(); + final List internalServices = new ArrayList<>(); + final List peers = new ArrayList<>(); + final List locations = new ArrayList<>(); + final List storageDirectories = new ArrayList<>(); + final List stateMachines = new ArrayList<>(); + final ConcurrentMap resetsReceived = new ConcurrentHashMap<>(); + try { + for (int nodeId = 1; nodeId <= 3; nodeId++) { + final LoopbackInternalService internal = + new LoopbackInternalService(nodeId, servers, resetsReceived); + internalServices.add(internal); + final TEndPoint consensusEndpoint = new TEndPoint("127.0.0.1", unusedPort()); + locations.add(location(nodeId, internal.getEndpoint(), consensusEndpoint)); + peers.add(new Peer(GROUP, nodeId, consensusEndpoint)); + storageDirectories.add(temporaryFolder.newFolder("peer-" + nodeId)); + stateMachines.add(new MemoryStateMachine()); + } + for (int index = 0; index < peers.size(); index++) { + final IoTConsensus server = + newServer(peers.get(index), storageDirectories.get(index), stateMachines.get(index)); + servers.put(peers.get(index).getNodeId(), server); + server.start(); + server.createLocalPeer(GROUP, peers); + } + final Peer writer = peers.get(0); + final Peer removed = peers.get(1); + final Peer coordinator = peers.get(2); + final List finalPeers = Arrays.asList(writer, coordinator); + final IoTConsensus originalWriter = servers.get(writer.getNodeId()); + write(originalWriter, INITIAL_WRITES); + await("initial replication", () -> originalWriter.getImpl(GROUP).getSyncLag() == 0); + final LogDispatcherThread originalRemovedThread = + dispatcher(originalWriter.getImpl(GROUP), removed); + + 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() + .removeRegionLocation(any(RemoveRegionLocationPlan.class))) + .thenAnswer(invocation -> partitionInfo.removeRegionLocation(invocation.getArgument(0))); + final RegionMaintainHandler handler = Mockito.spy(new RegionMaintainHandler(configManager)); + // These adapters replace asynchronous DN task submission, not the procedure's final reset. + Mockito.doNothing().when(handler).transferRegionLeader(any(), any(), any()); + Mockito.doAnswer( + invocation -> { + servers.get(coordinator.getNodeId()).removeRemotePeer(GROUP, removed); + return success(); + }) + .when(handler) + .submitRemoveRegionPeerTask(anyLong(), any(), any(), any()); + Mockito.doAnswer( + invocation -> { + servers.get(removed.getNodeId()).deleteLocalPeer(GROUP); + return success(); + }) + .when(handler) + .submitDeleteOldRegionPeerTask(anyLong(), any(), any()); + Mockito.doReturn(new TRegionMigrateResult(TRegionMaintainTaskStatus.SUCCESS)) + .when(handler) + .waitTaskFinish(anyLong(), any(TDataNodeLocation.class)); + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + Mockito.when(env.getRegionMaintainHandler()).thenReturn(handler); + final RemoveRegionPeerProcedure procedure = + new RemoveRegionPeerProcedure(REGION, locations.get(2), locations.get(1)); + procedure.executeFromState(env, RemoveRegionPeerState.TRANSFER_REGION_LEADER); + procedure.executeFromState(env, RemoveRegionPeerState.REMOVE_REGION_PEER); + assertEquals(finalPeers, originalWriter.getImpl(GROUP).getConfiguration()); + assertEquals( + Collections.singletonList(coordinator), dispatcherPeers(originalWriter.getImpl(GROUP))); + assertTrue(originalRemovedThread.isStopped()); + + // Obtain the same persisted replica-set source used for DataNode restart responses while + // the procedure is still between REMOVE_REGION_PEER and REMOVE_REGION_LOCATION_CACHE. + final List staleRestartLocations = + partitionInfo.getAllReplicaSets(writer.getNodeId()).get(0).getDataNodeLocations(); + assertTrue(staleRestartLocations.contains(locations.get(1))); + assertTrue(resetsReceived.isEmpty()); + procedure.executeFromState(env, RemoveRegionPeerState.DELETE_OLD_REGION_PEER); + procedure.executeFromState(env, RemoveRegionPeerState.DROP_CONSENSUS_PIPES); + originalWriter.stop(); + servers.remove(writer.getNodeId()); + final IoTConsensus restartedWriter = + newServer(writer, storageDirectories.get(0), stateMachines.get(0)); + servers.put(writer.getNodeId(), restartedWriter); + restartedWriter.recordCorrectPeerListBeforeStarting( + Collections.singletonMap(GROUP, toPeers(staleRestartLocations))); + restartedWriter.start(); + final IoTConsensusServerImpl recovered = restartedWriter.getImpl(GROUP); + assertEquals(peers, recovered.getConfiguration()); + final LogDispatcherThread recreatedRemovedThread = dispatcher(recovered, removed); + assertNotSame(originalRemovedThread, recreatedRemovedThread); + assertFalse(recreatedRemovedThread.isStopped()); + write(restartedWriter, WRITES_DURING_STALE_MEMBERSHIP); + final LogDispatcherThread healthyThread = dispatcher(recovered, coordinator); + final long expectedIndex = INITIAL_WRITES + WRITES_DURING_STALE_MEMBERSHIP; + await( + "healthy peer receives new writes", + () -> healthyThread.getCurrentSyncIndex() == expectedIndex); + await( + "removed peer fills its five-batch pipeline", + () -> { + synchronized (recreatedRemovedThread.getSyncStatus()) { + return recreatedRemovedThread.getSyncStatus().getPendingBatches().size() == 5; + } + }); + assertEquals(INITIAL_WRITES, recreatedRemovedThread.getCurrentSyncIndex()); + assertEquals(WRITES_DURING_STALE_MEMBERSHIP, recovered.getSyncLag()); + printState("STALE_RESTART_BEFORE_PROCEDURE_FINALIZATION", recovered); + + // Only this production state transition may trigger reset. The test body never invokes it. + procedure.executeFromState(env, RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE); + assertTrue(resetsReceived.isEmpty()); + procedure.executeFromState(env, RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST); + assertEquals( + new HashSet<>(Arrays.asList(writer.getNodeId(), coordinator.getNodeId())), + resetsReceived.keySet()); + for (TResetPeerListReq request : resetsReceived.values()) { + assertEquals(REGION, request.getRegionId()); + assertEquals(finalPeers, toPeers(request.getCorrectLocations())); + } + assertEquals(finalPeers, recovered.getConfiguration()); + assertEquals(Collections.singletonList(coordinator), dispatcherPeers(recovered)); + assertTrue(recreatedRemovedThread.isStopped()); + assertEquals(0, recovered.getSyncLag()); + assertFalse( + partitionInfo + .getAllReplicaSets(writer.getNodeId()) + .get(0) + .getDataNodeLocations() + .contains(locations.get(1))); + // The earlier response remains stale; correctness comes from the production reset RPC. + assertTrue(staleRestartLocations.contains(locations.get(1))); + printState("PROCEDURE_FINALIZATION_REPAIRED_MEMBERSHIP", recovered); + + write(restartedWriter, 2); + await("writes after procedure finalization", () -> recovered.getSyncLag() == 0); + assertEquals(expectedIndex + 2, stateMachines.get(2).values.size()); + printState("WRITES_CONTINUE_AFTER_REPAIR", recovered); + restartedWriter.stop(); + servers.remove(writer.getNodeId()); + final IoTConsensus freshWriter = + newServer(writer, storageDirectories.get(0), stateMachines.get(0)); + servers.put(writer.getNodeId(), freshWriter); + freshWriter.recordCorrectPeerListBeforeStarting( + Collections.singletonMap( + GROUP, + toPeers( + partitionInfo + .getAllReplicaSets(writer.getNodeId()) + .get(0) + .getDataNodeLocations()))); + freshWriter.start(); + assertEquals(finalPeers, freshWriter.getImpl(GROUP).getConfiguration()); + assertEquals( + Collections.singletonList(coordinator), dispatcherPeers(freshWriter.getImpl(GROUP))); + write(freshWriter, 1); + await( + "second restart uses committed membership", + () -> freshWriter.getImpl(GROUP).getSyncLag() == 0); + assertEquals(expectedIndex + 3, stateMachines.get(2).values.size()); + printState("SECOND_RESTART_DOES_NOT_RECREATE_REMOVED_PEER", freshWriter.getImpl(GROUP)); + } finally { + servers.values().parallelStream().forEach(IoTConsensus::stop); + for (LoopbackInternalService internal : internalServices) { + internal.close(); + } + cnConfig.setDataRegionConsensusProtocolClass(originalProtocol); + } + } + + private static PartitionInfo newPartitionInfo(List locations) { + final PartitionInfo partitionInfo = new PartitionInfo(); + partitionInfo.createDatabase( + new DatabaseSchemaPlan( + ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.restart_repair"))); + final CreateRegionGroupsPlan groups = new CreateRegionGroupsPlan(); + groups.addRegionGroup( + "root.restart_repair", new TRegionReplicaSet(REGION, new ArrayList<>(locations))); + partitionInfo.createRegionGroups(groups); + return partitionInfo; + } + + private static TDataNodeLocation location(int nodeId, TEndPoint internal, TEndPoint consensus) { + return new TDataNodeLocation( + nodeId, + new TEndPoint("127.0.0.1", 6667), + internal, + new TEndPoint("127.0.0.1", 10740), + consensus, + new TEndPoint("127.0.0.1", 10760)); + } + + private static List toPeers(List locations) { + return locations.stream() + .map(location -> Peer.valueOf(REGION, location)) + .sorted() + .collect(Collectors.toList()); + } + + private static IoTConsensus newServer( + Peer peer, File directory, MemoryStateMachine stateMachine) { + return new IoTConsensus( + ConsensusConfig.newBuilder() + .setThisNodeId(peer.getNodeId()) + .setThisNode(peer.getEndpoint()) + .setStorageDir(directory.getAbsolutePath()) + .setConsensusGroupType(TConsensusGroupType.DataRegion) + .setIoTConsensusConfig( + IoTConsensusConfig.newBuilder() + .setRpc( + IoTConsensusConfig.RPC.newBuilder().setConnectionTimeoutInMs(500).build()) + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setCheckpointGap(1) + .setMaxLogEntriesNumPerBatch(1) + .setMaxPendingBatchesNum(5) + .setBasicRetryWaitTimeMs(100) + .setMaxRetryWaitTimeMs(500) + .build()) + .build()) + .build(), + ignored -> stateMachine); + } + + private static TSStatus success() { + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + + private static int unusedPort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } + + private static void write(IoTConsensus server, int count) throws Exception { + final int previousCount = (int) server.getImpl(GROUP).getSearchIndex(); + for (int index = 1; index <= count; index++) { + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + server.write(GROUP, new IntegerRequest(previousCount + index)).getCode()); + } + } + + private static void await(String description, BooleanSupplier condition) throws Exception { + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(25); + while (!condition.getAsBoolean() && System.nanoTime() < deadline) { + Thread.sleep(20); + } + assertTrue(description, condition.getAsBoolean()); + } + + @SuppressWarnings("unchecked") + private static List dispatchers(IoTConsensusServerImpl server) + throws Exception { + final LogDispatcher dispatcher = server.getLogDispatcher(); + final Field field = LogDispatcher.class.getDeclaredField("threads"); + field.setAccessible(true); + synchronized (dispatcher) { + return new ArrayList<>((List) field.get(dispatcher)); + } + } + + private static List dispatcherPeers(IoTConsensusServerImpl server) throws Exception { + return dispatchers(server).stream() + .map(LogDispatcherThread::getPeer) + .collect(Collectors.toList()); + } + + private static LogDispatcherThread dispatcher(IoTConsensusServerImpl server, Peer peer) + throws Exception { + return dispatchers(server).stream() + .filter(thread -> thread.getPeer().equals(peer)) + .findFirst() + .get(); + } + + private static void printState(String stage, IoTConsensusServerImpl server) throws Exception { + System.out.printf( + "CN_REPAIR %s searchIndex=%d minSyncIndex=%d syncLag=%d dispatchers=%s%n", + stage, + server.getSearchIndex(), + server.getMinSyncIndex(), + server.getSyncLag(), + dispatchers(server).stream() + .map(thread -> thread.getPeer().getNodeId() + ":" + thread.getCurrentSyncIndex()) + .collect(Collectors.toList())); + } + + /** A real Thrift endpoint forwarding only resetPeerList to the selected consensus instance. */ + private static class LoopbackInternalService implements AutoCloseable { + private final TEndPoint endpoint; + private final ExecutorService workers; + private final TThreadPoolServer service; + private final Thread serverThread; + + private LoopbackInternalService( + int nodeId, + Map servers, + Map resetsReceived) + throws Exception { + final ServerSocket socket = new ServerSocket(0); + endpoint = new TEndPoint("127.0.0.1", socket.getLocalPort()); + final IDataNodeRPCService.Iface facade = Mockito.mock(IDataNodeRPCService.Iface.class); + Mockito.when(facade.resetPeerList(any(TResetPeerListReq.class))) + .thenAnswer( + invocation -> { + final TResetPeerListReq request = invocation.getArgument(0); + servers.get(nodeId).resetPeerList(GROUP, toPeers(request.getCorrectLocations())); + resetsReceived.put(nodeId, request.deepCopy()); + return success(); + }); + workers = + Executors.newCachedThreadPool( + runnable -> { + final Thread thread = new Thread(runnable, "reset-peer-facade-" + nodeId); + thread.setDaemon(true); + return thread; + }); + service = + new TThreadPoolServer( + new TThreadPoolServer.Args(new TServerSocket(socket, 1000)) + .processor(new IDataNodeRPCService.Processor<>(facade)) + .protocolFactory( + CommonDescriptor.getInstance().getConfig().isRpcThriftCompressionEnabled() + ? new TCompactProtocol.Factory() + : new TBinaryProtocol.Factory()) + .transportFactory(new TFramedTransport.Factory()) + .executorService(workers) + .stopTimeoutVal(1) + .stopTimeoutUnit(TimeUnit.SECONDS)); + serverThread = new Thread(service::serve, "reset-peer-server-" + nodeId); + serverThread.setDaemon(true); + serverThread.start(); + await("internal Thrift facade starts", service::isServing); + } + + private TEndPoint getEndpoint() { + return endpoint; + } + + @Override + public void close() throws Exception { + CnToDnInternalServiceAsyncRequestManager.getInstance().resetClient(endpoint); + service.stop(); + workers.shutdownNow(); + serverThread.join(TimeUnit.SECONDS.toMillis(3)); + } + } + + private static class IntegerRequest implements IConsensusRequest { + private final int value; + + private IntegerRequest(int value) { + this.value = value; + } + + @Override + public ByteBuffer serializeToByteBuffer() { + return ByteBuffer.allocate(Integer.BYTES).putInt(value).flip(); + } + } + + /** Keeps local WAL entries across consensus-instance restarts and records replicated values. */ + private static class MemoryStateMachine implements IStateMachine, ConsensusReqReader, DataSet { + private final Map localRequests = new HashMap<>(); + private final Set values = ConcurrentHashMap.newKeySet(); + private final AtomicLong localIndex = new AtomicLong(); + + @Override + public void start() {} + + @Override + public void stop() {} + + @Override + public TSStatus write(IConsensusRequest request) { + if (request instanceof DeserializedBatchIndexedConsensusRequest) { + final List statuses = new ArrayList<>(); + for (IConsensusRequest element : + ((DeserializedBatchIndexedConsensusRequest) request).getInsertNodes()) { + write(element); + statuses.add(success()); + } + return success().setSubStatus(statuses); + } + final IndexedConsensusRequest indexed = (IndexedConsensusRequest) request; + final List copied = new ArrayList<>(); + for (IConsensusRequest element : indexed.getRequests()) { + final int value = element.serializeToByteBuffer().duplicate().getInt(); + values.add(value); + copied.add(new IntegerRequest(value)); + } + if (indexed.getSearchIndex() != DEFAULT_SEARCH_INDEX) { + synchronized (localRequests) { + localRequests.put( + indexed.getSearchIndex(), + new IndexedConsensusRequest(indexed.getSearchIndex(), copied)); + localIndex.set(indexed.getSearchIndex()); + localRequests.notifyAll(); + } + } + return success(); + } + + @Override + public IConsensusRequest deserializeRequest(IConsensusRequest request) { + if (!(request instanceof BatchIndexedConsensusRequest)) { + return request; + } + final BatchIndexedConsensusRequest batch = (BatchIndexedConsensusRequest) request; + final DeserializedBatchIndexedConsensusRequest result = + new DeserializedBatchIndexedConsensusRequest( + batch.getStartSyncIndex(), + batch.getEndSyncIndex(), + batch.getRequests().size(), + batch.getSourcePeerId(), + 0); + batch.getRequests().forEach(result::add); + return result; + } + + @Override + public DataSet read(IConsensusRequest request) { + return request instanceof GetConsensusReqReaderPlan ? this : null; + } + + @Override + public boolean takeSnapshot(File directory) { + return true; + } + + @Override + public boolean loadSnapshot(File directory) { + return true; + } + + @Override + public void setSafelyDeletedSearchIndex(long index) {} + + @Override + public long getCurrentSearchIndex() { + return localIndex.get(); + } + + @Override + public long getCurrentWALFileVersion() { + return 0; + } + + @Override + public long getTotalSize() { + return 0; + } + + @Override + public ReqIterator getReqIterator(long startIndex) { + return new ReqIterator() { + private long nextIndex = startIndex; + + @Override + public boolean hasNext() { + synchronized (localRequests) { + return localRequests.containsKey(nextIndex); + } + } + + @Override + public IndexedConsensusRequest next() { + synchronized (localRequests) { + final IndexedConsensusRequest request = localRequests.get(nextIndex++); + return new IndexedConsensusRequest(request.getSearchIndex(), request.getRequests()); + } + } + + @Override + public void waitForNextReady() throws InterruptedException { + synchronized (localRequests) { + while (!hasNext()) { + localRequests.wait(); + } + } + } + + @Override + public void waitForNextReady(long timeout, TimeUnit unit) + throws InterruptedException, TimeoutException { + final long deadline = System.nanoTime() + unit.toNanos(timeout); + synchronized (localRequests) { + while (!hasNext()) { + final long remaining = deadline - System.nanoTime(); + if (remaining <= 0) { + throw new TimeoutException(); + } + TimeUnit.NANOSECONDS.timedWait(localRequests, remaining); + } + } + } + + @Override + public void skipTo(long targetIndex) { + nextIndex = targetIndex; + } + }; + } + } +} 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 new file mode 100644 index 0000000000000..be868283edcc8 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java @@ -0,0 +1,450 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.confignode.procedure.impl.region; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TRegionMaintainTaskStatus; +import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.cluster.RegionStatus; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType; +import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan; +import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan; +import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.manager.partition.PartitionManager; +import org.apache.iotdb.confignode.persistence.partition.PartitionInfo; +import org.apache.iotdb.confignode.procedure.Procedure; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; +import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler; +import org.apache.iotdb.confignode.procedure.state.ProcedureState; +import org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState; +import org.apache.iotdb.confignode.procedure.store.ProcedureType; +import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema; +import org.apache.iotdb.mpp.rpc.thrift.TRegionMigrateResult; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.Mockito; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.nio.ByteBuffer; +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 static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS; +import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS_V2; +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.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** Regression coverage for the stale restart-list window and bounded CN reconciliation. */ +public class RestartDuringRemovePeerTest { + private final TConsensusGroupId regionId = + new TConsensusGroupId(TConsensusGroupType.DataRegion, 6); + private final TDataNodeLocation restartingNode = location(3, "11.101.10.8"); + private final TDataNodeLocation coordinator = location(5, "11.101.10.9"); + private final TDataNodeLocation removedNode = location(6, "11.101.10.10"); + private final List finalPeers = Arrays.asList(restartingNode, coordinator); + private PartitionInfo partitionInfo; + private PartitionManager partitionManager; + private LoadManager loadManager; + private RegionMaintainHandler handler; + private ConfigNodeProcedureEnv env; + private String previousProtocol; + + @Before + public void setUp() throws Exception { + previousProtocol = + ConfigNodeDescriptor.getInstance().getConf().getDataRegionConsensusProtocolClass(); + ConfigNodeDescriptor.getInstance().getConf().setDataRegionConsensusProtocolClass(IOT_CONSENSUS); + partitionInfo = new PartitionInfo(); + partitionInfo.createDatabase( + new DatabaseSchemaPlan( + ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.restart"))); + CreateRegionGroupsPlan groups = new CreateRegionGroupsPlan(); + groups.addRegionGroup( + "root.restart", + new TRegionReplicaSet( + regionId, new ArrayList<>(Arrays.asList(restartingNode, coordinator, removedNode)))); + partitionInfo.createRegionGroups(groups); + + ConfigManager configManager = Mockito.mock(ConfigManager.class); + partitionManager = Mockito.mock(PartitionManager.class); + 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.removeRegionLocation(any())) + .thenAnswer(call -> partitionInfo.removeRegionLocation(call.getArgument(0))); + handler = Mockito.spy(new RegionMaintainHandler(configManager)); + env = Mockito.mock(ConfigNodeProcedureEnv.class); + when(env.getRegionMaintainHandler()).thenReturn(handler); + doNothing().when(handler).forceUpdateRegionCache(any(), any(), any()); + doNothing().when(handler).transferRegionLeader(any(), any(), any()); + doNothing().when(handler).dropConsensusPipesForRemovePeer(any(), any()); + doReturn(success()).when(handler).submitRemoveRegionPeerTask(anyLong(), any(), any(), any()); + doReturn(success()).when(handler).submitDeleteOldRegionPeerTask(anyLong(), any(), any()); + doReturn(new TRegionMigrateResult(TRegionMaintainTaskStatus.SUCCESS)) + .when(handler) + .waitTaskFinish(anyLong(), any(TDataNodeLocation.class)); + doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap()); + } + + @After + public void restoreProtocol() { + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(previousProtocol); + } + + @Test + public void staleRestartSnapshotIsFollowedByFinalReconciliation() throws Exception { + AccessibleRemoveProcedure procedure = newProcedure(); + procedure.doExecute(env); + procedure.doExecute(env); + List startupSnapshot = + partitionInfo.getAllReplicaSets(restartingNode.getDataNodeId()); + assertTrue(startupSnapshot.get(0).getDataNodeLocations().contains(removedNode)); + verify(handler, Mockito.atLeastOnce()) + .forceUpdateRegionCache(regionId, removedNode, RegionStatus.Removing); + verify(handler, never()).removeRegionLocation(regionId, removedNode); + procedure.doExecute(env); + procedure.doExecute(env); + + doAnswer( + call -> { + assertEquals( + finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations()); + assertEquals(finalPeers, call.getArgument(1)); + Map recipients = call.getArgument(2); + assertEquals(2, recipients.size()); + assertEquals(restartingNode, recipients.get(restartingNode.getDataNodeId())); + assertEquals(coordinator, recipients.get(coordinator.getDataNodeId())); + assertFalse(recipients.containsKey(removedNode.getDataNodeId())); + return successfulResponses(); + }) + .when(handler) + .resetPeerList(eq(regionId), anyList(), anyMap()); + + assertEquals(1, procedure.doExecute(env).length); + assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, procedure.currentState()); + verify(handler, never()).resetPeerList(any(), anyList(), anyMap()); + assertEquals(0, procedure.doExecute(env).length); + assertFalse(procedure.isWaiting()); + // Correcting future startup responses cannot change a snapshot already returned to a DN. + assertTrue(startupSnapshot.get(0).getDataNodeLocations().contains(removedNode)); + verify(handler).resetPeerList(eq(regionId), eq(finalPeers), anyMap()); + } + + @Test + public void metadataFailureRetriesBeforeCorrectingPeers() throws Exception { + AccessibleRemoveProcedure procedure = beforeMetadataRemoval(); + doReturn(new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode())) + .when(partitionManager) + .removeRegionLocation(any()); + assertEquals(1, procedure.doExecute(env).length); + assertEquals(ProcedureState.RUNNABLE, procedure.getState()); + assertEquals(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE, procedure.currentState()); + assertTrue( + partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations().contains(removedNode)); + verify(handler, never()).resetPeerList(any(), anyList(), anyMap()); + verify(loadManager).removeRegionCache(regionId, removedNode.getDataNodeId()); + verify(loadManager.getRouteBalancer()).balanceRegionLeaderAndPriority(); + + doAnswer(call -> partitionInfo.removeRegionLocation(call.getArgument(0))) + .when(partitionManager) + .removeRegionLocation(any()); + assertEquals(1, procedure.doExecute(env).length); + assertEquals(0, procedure.doExecute(env).length); + assertFalse(procedure.isWaiting()); + verify(handler).resetPeerList(eq(regionId), eq(finalPeers), anyMap()); + verify(loadManager, times(2)).removeRegionCache(regionId, removedNode.getDataNodeId()); + verify(loadManager.getRouteBalancer(), times(2)).balanceRegionLeaderAndPriority(); + } + + @Test + public void v2MetadataFailureRetriesBeforeCompletingWithoutResettingPeers() throws Exception { + ConfigNodeDescriptor.getInstance() + .getConf() + .setDataRegionConsensusProtocolClass(IOT_CONSENSUS_V2); + AccessibleRemoveProcedure procedure = beforeMetadataRemoval(); + doReturn(new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode())) + .doAnswer(call -> partitionInfo.removeRegionLocation(call.getArgument(0))) + .when(partitionManager) + .removeRegionLocation(any()); + + assertEquals(1, procedure.doExecute(env).length); + assertEquals(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE, procedure.currentState()); + assertFalse(procedure.isFailed()); + assertTrue( + partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations().contains(removedNode)); + + assertEquals(0, procedure.doExecute(env).length); + assertFalse(procedure.isFailed()); + assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations()); + verify(handler, times(2)).removeRegionLocation(regionId, removedNode); + verify(handler, never()).resetPeerList(any(), anyList(), anyMap()); + } + + @Test + public void missingAndNotInitializedResponsesCanRetryAfterRecovery() throws Exception { + AccessibleRemoveProcedure procedure = beforePeerListReset(); + doReturn(Collections.singletonMap(coordinator.getDataNodeId(), success())) + .when(handler) + .resetPeerList(any(), anyList(), anyMap()); + assertEquals(1, procedure.doExecute(env).length); + assertEquals(ProcedureState.RUNNABLE, procedure.getState()); + assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations()); + + AccessibleRemoveProcedure restored = restore(serialize(procedure)); + assertEquals(ProcedureState.RUNNABLE, restored.getState()); + assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, restored.currentState()); + Map notReady = successfulResponses(); + notReady.put( + restartingNode.getDataNodeId(), + new TSStatus(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode())); + doReturn(notReady).when(handler).resetPeerList(any(), anyList(), anyMap()); + assertEquals(1, restored.doExecute(env).length); + assertEquals(ProcedureState.RUNNABLE, restored.getState()); + + doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap()); + assertEquals(0, restored.doExecute(env).length); + assertFalse(restored.isWaiting()); + verify(handler, times(3)).resetPeerList(eq(regionId), eq(finalPeers), anyMap()); + verify(handler).removeRegionLocation(regionId, removedNode); + } + + @Test + public void exceptionDuringCorrectionIsRetriedInsteadOfCompleting() throws Exception { + AccessibleRemoveProcedure procedure = beforePeerListReset(); + doThrow(new IllegalStateException("injected RPC failure")) + .when(handler) + .resetPeerList(any(), anyList(), anyMap()); + assertEquals(1, procedure.doExecute(env).length); + assertEquals(ProcedureState.RUNNABLE, procedure.getState()); + assertFalse(procedure.isFailed()); + doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap()); + assertEquals(0, procedure.doExecute(env).length); + assertFalse(procedure.isWaiting()); + } + + @Test + public void replayAfterSuccessfulCorrectionIsIdempotent() throws Exception { + AccessibleRemoveProcedure procedure = beforePeerListReset(); + byte[] checkpoint = serialize(procedure); + assertEquals(0, procedure.doExecute(env).length); + // Crash after successful RPCs, before persisting procedure completion. + AccessibleRemoveProcedure restored = restore(checkpoint); + assertEquals(0, restored.doExecute(env).length); + assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations()); + verify(handler, times(2)).resetPeerList(eq(regionId), eq(finalPeers), anyMap()); + } + + @Test + public void offlineReplicaFailsAfterThreeRetriesInsteadOfWaitingForever() throws Exception { + AccessibleRemoveProcedure procedure = beforePeerListReset(); + doReturn(Collections.singletonMap(coordinator.getDataNodeId(), success())) + .when(handler) + .resetPeerList(any(), anyList(), anyMap()); + for (int retry = 0; retry < 3; retry++) { + assertEquals(1, procedure.doExecute(env).length); + assertEquals(ProcedureState.RUNNABLE, procedure.getState()); + assertFalse(procedure.isFailed()); + } + assertEquals(0, procedure.doExecute(env).length); + assertTrue(procedure.isFailed()); + assertTrue(procedure.getException().getMessage().contains("4")); + verify(handler, times(4)).resetPeerList(eq(regionId), eq(finalPeers), anyMap()); + verify(handler).removeRegionLocation(regionId, removedNode); + } + + @Test + public void metadataFailureAlsoStopsAtRetryLimitWithoutResettingPeers() throws Exception { + AccessibleRemoveProcedure procedure = beforeMetadataRemoval(); + doReturn(null).when(partitionManager).removeRegionLocation(any()); + for (int retry = 0; retry < 3; retry++) { + assertEquals(1, procedure.doExecute(env).length); + assertFalse(procedure.isWaiting()); + } + assertEquals(0, procedure.doExecute(env).length); + assertTrue(procedure.isFailed()); + verify(handler, times(4)).removeRegionLocation(regionId, removedNode); + verify(handler, never()).resetPeerList(any(), anyList(), anyMap()); + } + + @Test + public void interruptedRetryRetainsItsStepForRecovery() throws Exception { + AccessibleRemoveProcedure procedure = beforePeerListReset(); + doReturn(Collections.emptyMap()).when(handler).resetPeerList(any(), anyList(), anyMap()); + try { + Thread.currentThread().interrupt(); + assertEquals(1, procedure.doExecute(env).length); + assertTrue(Thread.currentThread().isInterrupted()); + assertFalse(procedure.isFailed()); + } finally { + Thread.interrupted(); + } + AccessibleRemoveProcedure restored = restore(serialize(procedure)); + assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, restored.currentState()); + doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap()); + assertEquals(0, restored.doExecute(env).length); + } + + @Test + public void correctionRejectsStaleMetadataAndNullResponsesButAcceptsDeletedRegion() { + assertFalse(handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, removedNode)); + verify(handler, never()).resetPeerList(any(), anyList(), anyMap()); + 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()); + assertTrue(handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, removedNode)); + } + + @Test + public void correctionIsLimitedToIoTConsensusV1DataRegions() throws Exception { + // New states must be appended so existing persisted state IDs remain valid. + assertEquals(3, RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE.ordinal()); + assertEquals(4, RemoveRegionPeerState.DROP_CONSENSUS_PIPES.ordinal()); + assertEquals(5, RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST.ordinal()); + assertTrue(handler.isIoTConsensusV1DataRegion(regionId)); + assertFalse( + handler.isIoTConsensusV1DataRegion( + new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 6))); + for (String protocol : Arrays.asList(IOT_CONSENSUS_V2, RATIS_CONSENSUS)) { + ConfigNodeDescriptor.getInstance().getConf().setDataRegionConsensusProtocolClass(protocol); + assertFalse(handler.isIoTConsensusV1DataRegion(regionId)); + AccessibleRemoveProcedure procedure = beforeMetadataRemoval(); + assertEquals(0, procedure.doExecute(env).length); + } + verify(handler, never()).resetPeerList(any(), anyList(), anyMap()); + } + + private AccessibleRemoveProcedure newProcedure() { + AccessibleRemoveProcedure procedure = + new AccessibleRemoveProcedure(regionId, coordinator, removedNode); + procedure.setProcId(1); + procedure.setState(ProcedureState.RUNNABLE); + return procedure; + } + + private AccessibleRemoveProcedure beforeMetadataRemoval() throws Exception { + AccessibleRemoveProcedure procedure = newProcedure(); + for (int i = 0; i < 4; i++) { + assertEquals(1, procedure.doExecute(env).length); + } + return procedure; + } + + private AccessibleRemoveProcedure beforePeerListReset() throws Exception { + AccessibleRemoveProcedure procedure = beforeMetadataRemoval(); + assertEquals(1, procedure.doExecute(env).length); + assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, procedure.currentState()); + return procedure; + } + + private static byte[] serialize(AccessibleRemoveProcedure procedure) throws Exception { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + procedure.serialize(new DataOutputStream(bytes)); + return bytes.toByteArray(); + } + + private static AccessibleRemoveProcedure restore(byte[] bytes) throws Exception { + ByteBuffer buffer = ByteBuffer.wrap(bytes); + assertEquals(ProcedureType.REMOVE_REGION_PEER_PROCEDURE.getTypeCode(), buffer.getShort()); + AccessibleRemoveProcedure restored = new AccessibleRemoveProcedure(); + restored.deserialize(buffer); + return restored; + } + + private Map successfulResponses() { + Map responses = new HashMap<>(); + for (TDataNodeLocation peer : finalPeers) { + responses.put(peer.getDataNodeId(), success()); + } + return responses; + } + + private static TSStatus success() { + return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); + } + + private static class AccessibleRemoveProcedure extends RemoveRegionPeerProcedure { + private AccessibleRemoveProcedure() {} + + private AccessibleRemoveProcedure( + TConsensusGroupId group, TDataNodeLocation coordinator, TDataNodeLocation target) { + super(group, coordinator, target); + } + + @Override + public Procedure[] doExecute(ConfigNodeProcedureEnv env) + throws InterruptedException { + return super.doExecute(env); + } + + @Override + public void setState(ProcedureState state) { + super.setState(state); + } + + private RemoveRegionPeerState currentState() { + return getCurrentState(); + } + } + + private static TDataNodeLocation location(int id, String ip) { + return new TDataNodeLocation( + id, + new TEndPoint(ip, 6667), + new TEndPoint(ip, 10730), + new TEndPoint(ip, 10740), + new TEndPoint(ip, 10750), + new TEndPoint(ip, 10760)); + } +} diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StalePeerListRestartReproductionTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StalePeerListRestartReproductionTest.java new file mode 100644 index 0000000000000..4400d6665e73c --- /dev/null +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StalePeerListRestartReproductionTest.java @@ -0,0 +1,272 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.consensus.iot; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.config.ConsensusConfig; +import org.apache.iotdb.consensus.config.IoTConsensusConfig; +import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher; +import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher.LogDispatcherThread; +import org.apache.iotdb.consensus.iot.util.TestEntry; +import org.apache.iotdb.consensus.iot.util.TestStateMachine; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.lang.reflect.Field; +import java.net.ServerSocket; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.function.BooleanSupplier; +import java.util.stream.Collectors; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertTrue; + +/** Diagnostic reproduction: assertions describe the current faulty behavior, not a fix. */ +public class StalePeerListRestartReproductionTest { + + private static final DataRegionId GROUP = new DataRegionId(91); + private static final int INITIAL_WRITES = 8; + private static final int WRITES_AFTER_RESTART = 8; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + @Test + public void removedPeerIsRecreatedByStaleStartupListAndPinsRealSyncLag() throws Exception { + final List peers = new ArrayList<>(); + final List storageDirectories = new ArrayList<>(); + final List stateMachines = new ArrayList<>(); + final IoTConsensus[] servers = new IoTConsensus[3]; + for (int index = 0; index < servers.length; index++) { + peers.add(new Peer(GROUP, index + 1, new TEndPoint("127.0.0.1", unusedPort()))); + storageDirectories.add(temporaryFolder.newFolder("peer-" + index)); + stateMachines.add(new TestStateMachine()); + } + final Peer writer = peers.get(0); + final Peer removed = peers.get(1); + final Peer survivor = peers.get(2); + final List finalPeers = Arrays.asList(writer, survivor); + try { + for (int index = 0; index < servers.length; index++) { + servers[index] = + newServer(peers.get(index), storageDirectories.get(index), stateMachines.get(index)); + servers[index].start(); + servers[index].createLocalPeer(GROUP, peers); + } + final IoTConsensus originalWriter = servers[0]; + write(originalWriter, writer, INITIAL_WRITES); + await("initial replicas catch up", () -> originalWriter.getImpl(GROUP).getSyncLag() == 0); + final LogDispatcherThread originalRemovedThread = + dispatcher(originalWriter.getImpl(GROUP), removed); + assertEquals(INITIAL_WRITES, originalRemovedThread.getCurrentSyncIndex()); + printState("BEFORE_REMOVE", originalWriter.getImpl(GROUP)); + + // This is the real remove RPC broadcast used by removeRemotePeer. A and C acknowledge it. + originalWriter.getImpl(GROUP).notifyPeersToRemoveSyncLogChannel(removed); + assertEquals(finalPeers, originalWriter.getImpl(GROUP).getConfiguration()); + assertEquals(finalPeers, servers[2].getImpl(GROUP).getConfiguration()); + assertEquals( + Collections.singletonList(survivor), dispatcherPeers(originalWriter.getImpl(GROUP))); + assertTrue(originalRemovedThread.isStopped()); + assertEquals(0, runFinished(originalRemovedThread).getCount()); + final File regionDirectory = new File(originalWriter.getRegionDirFromConsensusGroupId(GROUP)); + assertEquals(0, regionDirectory.listFiles((dir, name) -> name.startsWith("2-")).length); + printState("REMOVE_ACKNOWLEDGED", originalWriter.getImpl(GROUP)); + + // Keep B's RPC endpoint alive: it now returns INTERNAL_SERVER_ERROR (region does not exist). + servers[1].deleteLocalPeer(GROUP); + originalWriter.stop(); + servers[0] = null; + + // Model a restart registration response obtained before ConfigNode commits removal. + final IoTConsensus restartedWriter = + newServer(writer, storageDirectories.get(0), stateMachines.get(0)); + servers[0] = restartedWriter; + restartedWriter.recordCorrectPeerListBeforeStarting(Collections.singletonMap(GROUP, peers)); + restartedWriter.start(); + final IoTConsensusServerImpl recovered = restartedWriter.getImpl(GROUP); + assertEquals(peers, recovered.getConfiguration()); + assertEquals(Arrays.asList(removed, survivor), dispatcherPeers(recovered)); + final LogDispatcherThread resurrectedThread = dispatcher(recovered, removed); + assertNotSame(originalRemovedThread, resurrectedThread); + assertFalse(resurrectedThread.isStopped()); + assertEquals(1, runFinished(resurrectedThread).getCount()); + final long removedPeerIndex = resurrectedThread.getCurrentSyncIndex(); + assertEquals(INITIAL_WRITES, removedPeerIndex); + assertEquals(1, regionDirectory.listFiles((dir, name) -> name.startsWith("2-")).length); + printState("STALE_STARTUP_RECREATED_REMOVED_PEER", recovered); + + write(restartedWriter, writer, WRITES_AFTER_RESTART); + final long expectedSearchIndex = INITIAL_WRITES + WRITES_AFTER_RESTART; + final LogDispatcherThread healthyThread = dispatcher(recovered, survivor); + await( + "surviving replica catches up", + () -> healthyThread.getCurrentSyncIndex() == expectedSearchIndex); + assertEquals(expectedSearchIndex, recovered.getSearchIndex()); + assertEquals(expectedSearchIndex, stateMachines.get(2).getRequestSet().size()); + for (int sample = 0; sample < 5; sample++) { + assertEquals(removedPeerIndex, resurrectedThread.getCurrentSyncIndex()); + assertEquals(WRITES_AFTER_RESTART, recovered.getSyncLag()); + assertFalse(resurrectedThread.isStopped()); + assertEquals(1, runFinished(resurrectedThread).getCount()); + printState("LAG_PINNED_SAMPLE_" + sample, recovered); + Thread.sleep(200); + } + + // Later process shutdown changes the transport error, but does not repair membership or lag. + servers[1].stop(); + servers[1] = null; + Thread.sleep(300); + assertEquals(removedPeerIndex, resurrectedThread.getCurrentSyncIndex()); + assertEquals(WRITES_AFTER_RESTART, recovered.getSyncLag()); + printState("REMOVED_ENDPOINT_DOWN_LAG_STILL_PINNED", recovered); + + // Control: explicit membership reconciliation removes the new thread and immediately unpins + // lag. + restartedWriter.resetPeerList(GROUP, finalPeers); + assertEquals(Collections.singletonList(survivor), dispatcherPeers(recovered)); + assertTrue(resurrectedThread.isStopped()); + assertEquals(0, recovered.getSyncLag()); + printState("EXPLICIT_RECONCILIATION_UNPINS_LAG", recovered); + + restartedWriter.stop(); + servers[0] = null; + final IoTConsensus freshListWriter = + newServer(writer, storageDirectories.get(0), stateMachines.get(0)); + servers[0] = freshListWriter; + freshListWriter.recordCorrectPeerListBeforeStarting( + Collections.singletonMap(GROUP, finalPeers)); + freshListWriter.start(); + assertEquals(finalPeers, freshListWriter.getImpl(GROUP).getConfiguration()); + assertEquals( + Collections.singletonList(survivor), dispatcherPeers(freshListWriter.getImpl(GROUP))); + write(freshListWriter, writer, 1); + await("fresh-list restart converges", () -> freshListWriter.getImpl(GROUP).getSyncLag() == 0); + assertEquals(expectedSearchIndex + 1, freshListWriter.getImpl(GROUP).getSearchIndex()); + printState("FRESH_STARTUP_LIST_CONVERGES", freshListWriter.getImpl(GROUP)); + } finally { + Arrays.stream(servers) + .filter(server -> server != null) + .parallel() + .forEach(IoTConsensus::stop); + } + } + + private static IoTConsensus newServer(Peer peer, File directory, TestStateMachine stateMachine) { + return new IoTConsensus( + ConsensusConfig.newBuilder() + .setThisNodeId(peer.getNodeId()) + .setThisNode(peer.getEndpoint()) + .setStorageDir(directory.getAbsolutePath()) + .setConsensusGroupType(TConsensusGroupType.DataRegion) + .setIoTConsensusConfig( + IoTConsensusConfig.newBuilder() + .setRpc( + IoTConsensusConfig.RPC.newBuilder().setConnectionTimeoutInMs(500).build()) + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setCheckpointGap(1) + .setBasicRetryWaitTimeMs(20) + .setMaxRetryWaitTimeMs(100) + .build()) + .build()) + .build(), + ignored -> stateMachine); + } + + private static int unusedPort() throws Exception { + try (ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } + + private static void write(IoTConsensus server, Peer writer, int count) throws Exception { + final int previousCount = (int) server.getImpl(GROUP).getSearchIndex(); + for (int index = 1; index <= count; index++) { + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + server.write(GROUP, new TestEntry(previousCount + index, writer)).getCode()); + } + } + + private static void await(String condition, BooleanSupplier satisfied) throws Exception { + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20); + while (!satisfied.getAsBoolean() && System.nanoTime() < deadline) { + Thread.sleep(20); + } + assertTrue(condition, satisfied.getAsBoolean()); + } + + @SuppressWarnings("unchecked") + private static List dispatchers(IoTConsensusServerImpl server) + throws Exception { + final LogDispatcher dispatcher = server.getLogDispatcher(); + final Field field = LogDispatcher.class.getDeclaredField("threads"); + field.setAccessible(true); + synchronized (dispatcher) { + return new ArrayList<>((List) field.get(dispatcher)); + } + } + + private static List dispatcherPeers(IoTConsensusServerImpl server) throws Exception { + return dispatchers(server).stream() + .map(LogDispatcherThread::getPeer) + .collect(Collectors.toList()); + } + + private static LogDispatcherThread dispatcher(IoTConsensusServerImpl server, Peer peer) + throws Exception { + return dispatchers(server).stream() + .filter(thread -> thread.getPeer().equals(peer)) + .findFirst() + .get(); + } + + private static CountDownLatch runFinished(LogDispatcherThread thread) throws Exception { + final Field field = LogDispatcherThread.class.getDeclaredField("runFinished"); + field.setAccessible(true); + return (CountDownLatch) field.get(thread); + } + + private static void printState(String stage, IoTConsensusServerImpl server) throws Exception { + System.out.printf( + "REPRO %s searchIndex=%d minSyncIndex=%d syncLag=%d dispatchers=%s%n", + stage, + server.getSearchIndex(), + server.getMinSyncIndex(), + server.getSyncLag(), + dispatchers(server).stream() + .map(thread -> thread.getPeer().getNodeId() + ":" + thread.getCurrentSyncIndex()) + .collect(Collectors.toList())); + } +}