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())); + } +}