This is an automated email from the ASF dual-hosted git repository.
CRZbulabula pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 847cc6d1848 Fix ConfigNode leader transfer retries during removal
(#18348)
847cc6d1848 is described below
commit 847cc6d1848cedb50d838378c3841d4b8afee0f9
Author: Yongzao <[email protected]>
AuthorDate: Wed Jul 29 17:15:06 2026 +0800
Fix ConfigNode leader transfer retries during removal (#18348)
---
.../IoTDBRemoveConfigNodeITFramework.java | 20 ++-
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../iotdb/confignode/i18n/ManagerMessages.java | 3 +
.../iotdb/confignode/manager/node/NodeManager.java | 79 +++++++--
.../confignode/manager/node/NodeManagerTest.java | 189 +++++++++++++++++++++
5 files changed, 268 insertions(+), 26 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removeconfignode/IoTDBRemoveConfigNodeITFramework.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removeconfignode/IoTDBRemoveConfigNodeITFramework.java
index 6112fea7f52..54552ad4867 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removeconfignode/IoTDBRemoveConfigNodeITFramework.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removeconfignode/IoTDBRemoveConfigNodeITFramework.java
@@ -19,11 +19,11 @@
package org.apache.iotdb.confignode.it.removeconfignode;
-import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant;
import org.apache.iotdb.confignode.it.removedatanode.SQLModel;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
import org.apache.iotdb.itbase.exception.InconsistentDataException;
import org.apache.iotdb.jdbc.IoTDBSQLException;
import org.apache.iotdb.relational.it.query.old.aligned.TableUtils;
@@ -97,9 +97,7 @@ public class IoTDBRemoveConfigNodeITFramework {
EnvFactory.getEnv().initClusterEnvironment(configNodeNum, dataNodeNum);
try (final Connection connection =
makeItCloseQuietly(getConnectionWithSQLType(model));
- final Statement statement =
makeItCloseQuietly(connection.createStatement());
- SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
+ final Statement statement =
makeItCloseQuietly(connection.createStatement())) {
if (SQLModel.TABLE_MODEL_SQL.equals(model)) {
// Insert data in table model
@@ -120,14 +118,18 @@ public class IoTDBRemoveConfigNodeITFramework {
// Get all config nodes
ResultSet result = statement.executeQuery(SHOW_CONFIGNODES);
- Set<Integer> allConfigNodeId = new HashSet<>();
+ int removeConfigNodeId = -1;
+ ConfigNodeWrapper leaderConfigNode =
+
EnvFactory.getEnv().getConfigNodeWrapper(EnvFactory.getEnv().getLeaderConfigNodeIndex());
while (result.next()) {
- allConfigNodeId.add(result.getInt(ColumnHeaderConstant.NODE_ID));
+ int configNodeId = result.getInt(ColumnHeaderConstant.NODE_ID);
+ if
(leaderConfigNode.getIp().equals(result.getString(ColumnHeaderConstant.INTERNAL_ADDRESS))
+ && leaderConfigNode.getPort() ==
result.getInt(ColumnHeaderConstant.INTERNAL_PORT)) {
+ removeConfigNodeId = configNodeId;
+ }
}
+ Assert.assertNotEquals(-1, removeConfigNodeId);
- AtomicReference<SyncConfigNodeIServiceClient> clientRef = new
AtomicReference<>(client);
-
- int removeConfigNodeId = allConfigNodeId.iterator().next();
String removeConfigNodeSQL = generateRemoveString(removeConfigNodeId);
LOGGER.info("Remove ConfigNodes SQL: {}", removeConfigNodeSQL);
try {
diff --git
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 7028e97be7e..13bfaf6de8a 100644
---
a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -551,6 +551,9 @@ public final class ManagerMessages {
public static final String
REMOVE_CONFIGNODE_FAILED_BECAUSE_THE_CONFIGNODE_NOT_IN_CURRENT_CLUSTER =
"Remove ConfigNode failed because the ConfigNode not in current Cluster.";
public static final String SUCCESSFULLY_REMOVE_CONFIGNODE = "Successfully
remove confignode.";
public static final String
REMOVE_CONFIGNODE_FAILED_BECAUSE_TRANSFER_CONFIGNODE_LEADER_FAILED = "Remove
ConfigNode failed because transfer ConfigNode leader failed.";
+ public static final String
LOG_FAILED_TO_TRANSFER_CONFIGNODE_LEADER_FROM_ARG_TO_ARG_TRYING_ANOTHER_CANDIDATE_BA922E92
= "Failed to transfer ConfigNode leader from {} to {}, trying another
candidate.";
+ public static final String
LOG_COULD_NOT_CONFIRM_A_CONFIGNODE_LEADER_OTHER_THAN_ARG_AFTER_ATTEMPTING_TO_TRANSFER_LEADERSHIP_TO_ARG_TRYING_ANOTHER_CANDIDATE_91EF68C1
= "Could not confirm a ConfigNode leader other than {} after attempting to
transfer leadership to {}, trying another candidate.";
+ public static final String
LOG_STOPPED_RETRYING_CONFIGNODE_LEADER_TRANSFER_BECAUSE_THE_REMAINING_RPC_TIMEOUT_IS_INSUFFICIENT_6429A49C
= "Stopped retrying ConfigNode leader transfer because the remaining RPC
timeout is insufficient.";
private ManagerMessages() {}
//
---------------------------------------------------------------------------
diff --git
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
index 77c005cddf4..42aa21f46ae 100644
---
a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
+++
b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java
@@ -546,6 +546,9 @@ public final class ManagerMessages {
public static final String
REMOVE_CONFIGNODE_FAILED_BECAUSE_THE_CONFIGNODE_NOT_IN_CURRENT_CLUSTER = "移除
ConfigNode 失败，该 ConfigNode 不在当前集群中。";
public static final String SUCCESSFULLY_REMOVE_CONFIGNODE = "成功移除
ConfigNode。";
public static final String
REMOVE_CONFIGNODE_FAILED_BECAUSE_TRANSFER_CONFIGNODE_LEADER_FAILED = "移除
ConfigNode 失败，转移 ConfigNode leader 失败。";
+ public static final String
LOG_FAILED_TO_TRANSFER_CONFIGNODE_LEADER_FROM_ARG_TO_ARG_TRYING_ANOTHER_CANDIDATE_BA922E92
= "将 ConfigNode leader 从 {} 转移到 {} 失败，将尝试其他候选节点。";
+ public static final String
LOG_COULD_NOT_CONFIRM_A_CONFIGNODE_LEADER_OTHER_THAN_ARG_AFTER_ATTEMPTING_TO_TRANSFER_LEADERSHIP_TO_ARG_TRYING_ANOTHER_CANDIDATE_91EF68C1
= "无法确认存在除 ConfigNode {} 以外的 leader；此前尝试的转移目标为 {}，将尝试其他候选节点。";
+ public static final String
LOG_STOPPED_RETRYING_CONFIGNODE_LEADER_TRANSFER_BECAUSE_THE_REMAINING_RPC_TIMEOUT_IS_INSUFFICIENT_6429A49C
= "剩余 RPC 超时时间不足，停止重试 ConfigNode leader 转移。";
private ManagerMessages() {}
//
---------------------------------------------------------------------------
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 293f1942ff5..64d4cf741e1 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -104,6 +104,7 @@ import
org.apache.iotdb.confignode.rpc.thrift.TSetDataNodeStatusReq;
import org.apache.iotdb.consensus.common.DataSet;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.exception.RatisRequestFailedException;
import org.apache.iotdb.mpp.rpc.thrift.TKillQueryInstanceReq;
import org.apache.iotdb.rpc.RpcUtils;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -985,30 +986,74 @@ public class NodeManager {
private TSStatus transferLeader(
RemoveConfigNodePlan removeConfigNodePlan, ConsensusGroupId groupId) {
- Optional<TConfigNodeLocation> optional =
+ final TConfigNodeLocation removedConfigNode =
removeConfigNodePlan.getConfigNodeLocation();
+ final List<TConfigNodeLocation> newLeaderCandidates =
filterConfigNodeThroughStatus(NodeStatus.Running).stream()
- .filter(e ->
!e.equals(removeConfigNodePlan.getConfigNodeLocation()))
- .findAny();
- TConfigNodeLocation newLeader = null;
- if (optional.isPresent()) {
- newLeader = optional.get();
- } else {
+ .filter(configNode -> !configNode.equals(removedConfigNode))
+
.sorted(Comparator.comparingInt(TConfigNodeLocation::getConfigNodeId))
+ .collect(Collectors.toList());
+ if (newLeaderCandidates.isEmpty()) {
return new TSStatus(TSStatusCode.TRANSFER_LEADER_ERROR.getStatusCode())
.setMessage(
ManagerMessages
.MESSAGE_TRANSFER_CONFIGNODE_LEADER_FAILED_BECAUSE_CAN_NOT_FIND_ANY_RUNNING_1FE4F96D);
}
- try {
- getConsensusManager()
- .getConsensusImpl()
- .transferLeader(
- groupId,
- new Peer(groupId, newLeader.getConfigNodeId(),
newLeader.getConsensusEndPoint()));
- } catch (ConsensusException e) {
- return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
- .setMessage(
-
ManagerMessages.REMOVE_CONFIGNODE_FAILED_BECAUSE_TRANSFER_CONFIGNODE_LEADER_FAILED);
+
+ final long retryDeadline =
+ System.nanoTime()
+ +
TimeUnit.MILLISECONDS.toNanos(COMMON_CONFIG.getCnConnectionTimeoutInMS());
+ final long transferLeaderTimeout =
+ TimeUnit.MILLISECONDS.toNanos(CONF.getRatisTransferLeaderTimeoutMs());
+
+ for (int i = 0; i < newLeaderCandidates.size(); i++) {
+ if (i > 0 && retryDeadline - System.nanoTime() <= transferLeaderTimeout)
{
+ LOGGER.warn(
+ ManagerMessages
+
.LOG_STOPPED_RETRYING_CONFIGNODE_LEADER_TRANSFER_BECAUSE_THE_REMAINING_RPC_TIMEOUT_IS_INSUFFICIENT_6429A49C);
+ break;
+ }
+
+ final TConfigNodeLocation newLeaderCandidate =
newLeaderCandidates.get(i);
+ try {
+ getConsensusManager()
+ .getConsensusImpl()
+ .transferLeader(
+ groupId,
+ new Peer(
+ groupId,
+ newLeaderCandidate.getConfigNodeId(),
+ newLeaderCandidate.getConsensusEndPoint()));
+ } catch (RatisRequestFailedException e) {
+ LOGGER.warn(
+ ManagerMessages
+
.LOG_FAILED_TO_TRANSFER_CONFIGNODE_LEADER_FROM_ARG_TO_ARG_TRYING_ANOTHER_CANDIDATE_BA922E92,
+ removedConfigNode.getConfigNodeId(),
+ newLeaderCandidate.getConfigNodeId(),
+ e);
+ } catch (ConsensusException e) {
+ return new
TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
+ .setMessage(
+
ManagerMessages.REMOVE_CONFIGNODE_FAILED_BECAUSE_TRANSFER_CONFIGNODE_LEADER_FAILED);
+ }
+
+ final TConfigNodeLocation actualLeader =
getConsensusManager().getLeaderLocation();
+ if (actualLeader != null && !actualLeader.equals(removedConfigNode)) {
+ return buildConfigNodeLeaderTransferredStatus(actualLeader);
+ }
+
+ LOGGER.warn(
+ ManagerMessages
+
.LOG_COULD_NOT_CONFIRM_A_CONFIGNODE_LEADER_OTHER_THAN_ARG_AFTER_ATTEMPTING_TO_TRANSFER_LEADERSHIP_TO_ARG_TRYING_ANOTHER_CANDIDATE_91EF68C1,
+ removedConfigNode.getConfigNodeId(),
+ newLeaderCandidate.getConfigNodeId());
}
+
+ return new TSStatus(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode())
+ .setMessage(
+
ManagerMessages.REMOVE_CONFIGNODE_FAILED_BECAUSE_TRANSFER_CONFIGNODE_LEADER_FAILED);
+ }
+
+ private TSStatus buildConfigNodeLeaderTransferredStatus(TConfigNodeLocation
newLeader) {
return new TSStatus(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode())
.setRedirectNode(newLeader.getInternalEndPoint())
.setMessage(
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/node/NodeManagerTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/node/NodeManagerTest.java
new file mode 100644
index 00000000000..f39ecd46d5c
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/node/NodeManagerTest.java
@@ -0,0 +1,189 @@
+/*
+ * 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.manager.node;
+
+import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import
org.apache.iotdb.confignode.consensus.request.write.confignode.RemoveConfigNodePlan;
+import org.apache.iotdb.confignode.manager.IManager;
+import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
+import org.apache.iotdb.confignode.manager.load.LoadManager;
+import org.apache.iotdb.confignode.persistence.node.NodeInfo;
+import org.apache.iotdb.consensus.IConsensus;
+import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.consensus.exception.ConsensusException;
+import org.apache.iotdb.consensus.exception.RatisRequestFailedException;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class NodeManagerTest {
+
+ private final TConfigNodeLocation removedConfigNode =
newConfigNodeLocation(0);
+ private final TConfigNodeLocation firstCandidate = newConfigNodeLocation(1);
+ private final TConfigNodeLocation secondCandidate = newConfigNodeLocation(2);
+ private final List<TConfigNodeLocation> configNodes =
+ Arrays.asList(removedConfigNode, firstCandidate, secondCandidate);
+
+ private IConsensus consensus;
+ private ConsensusManager consensusManager;
+ private NodeManager nodeManager;
+ private int originalCnConnectionTimeout;
+ private int originalTransferLeaderTimeout;
+
+ @Before
+ public void setUp() {
+ originalCnConnectionTimeout =
+
CommonDescriptor.getInstance().getConfig().getCnConnectionTimeoutInMS();
+ originalTransferLeaderTimeout =
+
ConfigNodeDescriptor.getInstance().getConf().getRatisTransferLeaderTimeoutMs();
+
CommonDescriptor.getInstance().getConfig().setCnConnectionTimeoutInMS(60_000);
+
ConfigNodeDescriptor.getInstance().getConf().setRatisTransferLeaderTimeoutMs(1_000);
+
+ consensus = Mockito.mock(IConsensus.class);
+ consensusManager = Mockito.mock(ConsensusManager.class);
+ IManager configManager = Mockito.mock(IManager.class);
+ NodeInfo nodeInfo = Mockito.mock(NodeInfo.class);
+ LoadManager loadManager = Mockito.mock(LoadManager.class);
+
+
Mockito.when(configManager.getConsensusManager()).thenReturn(consensusManager);
+ Mockito.when(configManager.getLoadManager()).thenReturn(loadManager);
+ Mockito.when(consensusManager.getConsensusImpl()).thenReturn(consensus);
+ Mockito.when(consensusManager.getConsensusGroupId())
+ .thenReturn(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID);
+ Mockito.when(nodeInfo.getRegisteredConfigNodes()).thenReturn(configNodes);
+
Mockito.when(loadManager.filterConfigNodeThroughStatus(Mockito.any(NodeStatus[].class)))
+ .thenReturn(
+ configNodes.stream()
+ .map(TConfigNodeLocation::getConfigNodeId)
+ .collect(Collectors.toList()));
+
Mockito.when(nodeInfo.getRegisteredConfigNodes(Mockito.anyList())).thenReturn(configNodes);
+
+ nodeManager = new NodeManager(configManager, nodeInfo);
+ }
+
+ @After
+ public void tearDown() {
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setCnConnectionTimeoutInMS(originalCnConnectionTimeout);
+ ConfigNodeDescriptor.getInstance()
+ .getConf()
+ .setRatisTransferLeaderTimeoutMs(originalTransferLeaderTimeout);
+ }
+
+ @Test
+ public void transferLeaderShouldTryAnotherCandidateAfterTransientFailure()
throws Exception {
+ Mockito.when(consensusManager.getLeaderLocation())
+ .thenReturn(removedConfigNode, removedConfigNode, secondCandidate);
+ Mockito.doThrow(new RatisRequestFailedException(new Exception("transfer
failed")))
+ .doNothing()
+ .when(consensus)
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
Mockito.any(Peer.class));
+
+ TSStatus status =
+ nodeManager.checkConfigNodeBeforeRemove(new
RemoveConfigNodePlan(removedConfigNode));
+
+ Assert.assertEquals(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode(),
status.getCode());
+ Assert.assertEquals(secondCandidate.getInternalEndPoint(),
status.getRedirectNode());
+ ArgumentCaptor<Peer> peerCaptor = ArgumentCaptor.forClass(Peer.class);
+ Mockito.verify(consensus, Mockito.times(2))
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
peerCaptor.capture());
+ Assert.assertEquals(
+ Arrays.asList(firstCandidate.getConfigNodeId(),
secondCandidate.getConfigNodeId()),
+
peerCaptor.getAllValues().stream().map(Peer::getNodeId).collect(Collectors.toList()));
+ }
+
+ @Test
+ public void transferLeaderShouldRedirectToActualLeaderAfterFailedResponse()
throws Exception {
+ Mockito.when(consensusManager.getLeaderLocation())
+ .thenReturn(removedConfigNode, secondCandidate);
+ Mockito.doThrow(new RatisRequestFailedException(new Exception("transfer
failed")))
+ .when(consensus)
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
Mockito.any(Peer.class));
+
+ TSStatus status =
+ nodeManager.checkConfigNodeBeforeRemove(new
RemoveConfigNodePlan(removedConfigNode));
+
+ Assert.assertEquals(TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode(),
status.getCode());
+ Assert.assertEquals(secondCandidate.getInternalEndPoint(),
status.getRedirectNode());
+ Mockito.verify(consensus)
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
Mockito.any(Peer.class));
+ }
+
+ @Test
+ public void transferLeaderShouldFailAfterAllCandidatesFail() throws
Exception {
+ Mockito.when(consensusManager.getLeaderLocation())
+ .thenReturn(removedConfigNode, removedConfigNode, removedConfigNode);
+ Mockito.doThrow(new RatisRequestFailedException(new Exception("transfer
failed")))
+ .when(consensus)
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
Mockito.any(Peer.class));
+
+ TSStatus status =
+ nodeManager.checkConfigNodeBeforeRemove(new
RemoveConfigNodePlan(removedConfigNode));
+
+ Assert.assertEquals(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode(),
status.getCode());
+ Mockito.verify(consensus, Mockito.times(2))
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
Mockito.any(Peer.class));
+ }
+
+ @Test
+ public void transferLeaderShouldNotRetryNonRatisFailure() throws Exception {
+
Mockito.when(consensusManager.getLeaderLocation()).thenReturn(removedConfigNode);
+ Mockito.doThrow(new ConsensusException("non-retriable failure"))
+ .when(consensus)
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
Mockito.any(Peer.class));
+
+ TSStatus status =
+ nodeManager.checkConfigNodeBeforeRemove(new
RemoveConfigNodePlan(removedConfigNode));
+
+ Assert.assertEquals(TSStatusCode.REMOVE_CONFIGNODE_ERROR.getStatusCode(),
status.getCode());
+ Mockito.verify(consensus)
+ .transferLeader(
+ Mockito.eq(ConsensusManager.DEFAULT_CONSENSUS_GROUP_ID),
Mockito.any(Peer.class));
+ }
+
+ private static TConfigNodeLocation newConfigNodeLocation(int configNodeId) {
+ return new TConfigNodeLocation(
+ configNodeId,
+ new TEndPoint("127.0.0.1", 10710 + configNodeId),
+ new TEndPoint("127.0.0.1", 10720 + configNodeId));
+ }
+}
