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 e4090f7c740 Verify node status failover through production RPCs
(#18963)
e4090f7c740 is described below
commit e4090f7c740e32494ba52d893959e3889b6767bd
Author: d-wang-commit <[email protected]>
AuthorDate: Fri Oct 23 16:01:43 2026 +0800
Verify node status failover through production RPCs (#18963)
* Reuse the JDBC session in the node status failover test
Open the healthy DataNode session before injecting failures and reuse it
for SHOW assertions.
Preserve status and quorum checks and use the configured default
credentials.
* Stabilize node status failover integration tests
Reuse pre-fault sessions and wait for persistence and process exit.
Use A -> B -> C -> A with one fresh-cluster retry, verify SQL after quorum
recovery, and remove a DataNode hosting the tested database.
* Retry transient JVM attachment failures in node status tests
Retry the attach handshake once while the ConfigNode process is alive.
Keep test commands outside the retry and preserve both failures.
* Verify node status failover through production RPCs
Replace JVM instrumentation with production RPCs and process failures.
Limit assertions to observable status recovery and registration outcomes.
---
.../it/cluster/IoTDBNodeStatusFailoverIT.java | 836 ++++++---------------
.../confignode/it/cluster/NodeStatusTestAgent.java | 203 -----
2 files changed, 241 insertions(+), 798 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
index 13dd39c7a31..b9680597596 100644
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBNodeStatusFailoverIT.java
@@ -21,684 +21,340 @@ package org.apache.iotdb.confignode.it.cluster;
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.client.ClientPoolFactory;
import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.exception.ClientManagerException;
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
import org.apache.iotdb.commons.client.sync.SyncDataNodeInternalServiceClient;
import org.apache.iotdb.commons.cluster.NodeStatus;
-import org.apache.iotdb.confignode.persistence.node.NodeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TConfigNodeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveReq;
import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowConfigNodesResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.AbstractNodeWrapper;
+import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
import org.apache.iotdb.it.framework.IoTDBTestRunner;
import org.apache.iotdb.itbase.category.ClusterIT;
import org.apache.iotdb.rpc.TSStatusCode;
-import com.sun.tools.attach.VirtualMachine;
-import org.apache.tsfile.utils.Pair;
+import org.apache.thrift.TException;
import org.junit.After;
-import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;
-import org.junit.rules.ErrorCollector;
-import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.UncheckedIOException;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Properties;
import java.util.concurrent.TimeUnit;
-import java.util.jar.Attributes;
-import java.util.jar.JarEntry;
-import java.util.jar.JarOutputStream;
-import java.util.jar.Manifest;
import java.util.stream.Collectors;
-import java.util.stream.Stream;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-/** Process-level status persistence checks with three actual ConfigRegion
voters. */
+/** Node status recovery observed through production ConfigNode RPCs and
actual process failures. */
@RunWith(IoTDBTestRunner.class)
@Category(ClusterIT.class)
public class IoTDBNodeStatusFailoverIT {
- @Rule public TemporaryFolder temporary = new TemporaryFolder();
- @Rule public ErrorCollector failures = new ErrorCollector();
- private Path agentJar;
- private int queryNode;
+ private IClientManager<TEndPoint, SyncConfigNodeIServiceClient>
configNodeClients;
@After
- public void tearDown() {
- EnvFactory.getEnv().cleanClusterEnvironment();
- }
-
- @Test
- public void testGracefulLeaderShutdownPreservesStoppedAndRemoving() throws
Exception {
- init(3);
- int stopped = stoppedFixture(0);
- int removing = removingFixture(1);
- int leader = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- Path marker = submit(leader, "shutdown");
- awaitFile(marker);
- assertTrue(
- "System.exit must finish all JVM shutdown hooks",
- EnvFactory.getEnv()
- .getConfigNodeWrapper(leader)
- .getInstance()
- .waitFor(60, TimeUnit.SECONDS));
- assertNotEquals(leader, EnvFactory.getEnv().getLeaderConfigNodeIndex());
- assertStatus(0, stopped, NodeStatus.Stopped);
- assertStatus(1, removing, NodeStatus.Removing);
- EnvFactory.getEnv().startConfigNode(leader);
- awaitConfigNodeRunning(leader);
- awaitPersisted(leader, stopped, NodeStatus.Stopped);
- transfer(leader);
- assertStatus(0, stopped, NodeStatus.Stopped);
- assertStatus(1, removing, NodeStatus.Removing);
+ public void tearDown() throws InterruptedException {
+ List<AbstractNodeWrapper> nodes = new
ArrayList<>(EnvFactory.getEnv().getNodeWrapperList());
+ try {
+ // Stop every process before directory cleanup can fail and skip the
remaining nodes.
+ for (AbstractNodeWrapper node : nodes) {
+ node.stopForcibly();
+ }
+ for (AbstractNodeWrapper node : nodes) {
+ if (node.getInstance() != null) {
+ assertTrue(node.getId(), node.getInstance().waitFor(60,
TimeUnit.SECONDS));
+ assertFalse(node.getId(), node.isAlive());
+ }
+ }
+ } finally {
+ try {
+ if (configNodeClients != null) {
+ configNodeClients.close();
+ }
+ } finally {
+ EnvFactory.getEnv().cleanClusterEnvironment();
+ }
+ }
}
@Test
- public void testGracefulSeedShutdownReportsStopped() throws Exception {
+ public void testReportedConfigNodeShutdownClearsAfterRestart() throws
Exception {
init(1);
- transfer(1);
- int seedId = configLocation(cluster(), 0).getConfigNodeId();
- // Use System.exit so the actual shutdown hook runs on Windows as well as
Unix.
- awaitFile(submit(0, "shutdown"));
- assertTrue(
- EnvFactory.getEnv().getConfigNodeWrapper(0).getInstance().waitFor(60,
TimeUnit.SECONDS));
- EnvFactory.getEnv()
- .ensureNodeStatus(
-
Collections.singletonList(EnvFactory.getEnv().getConfigNodeWrapper(0)),
- Collections.singletonList(NodeStatus.Stopped));
- awaitPersisted(1, seedId, NodeStatus.Stopped);
- awaitPersisted(2, seedId, NodeStatus.Stopped);
+ int follower = (awaitLeader() + 1) % 3;
+ TConfigNodeLocation location = configLocation(cluster(), follower);
+ stopForcibly(EnvFactory.getEnv().getConfigNodeWrapper(follower));
+ // Exercise the production shutdown report, without depending on
platform-specific JVM hooks.
+ try (SyncConfigNodeIServiceClient client = leaderClient()) {
+ assertSuccess(client.reportConfigNodeShutdown(location));
+ }
+ awaitConfigNodeStatus(location.getConfigNodeId(), NodeStatus.Stopped);
- EnvFactory.getEnv().startConfigNode(0);
- awaitConfigNodeRunning(0);
- awaitPersisted(0, seedId, null);
- awaitPersisted(1, seedId, null);
- awaitPersisted(2, seedId, null);
+ EnvFactory.getEnv().startConfigNode(follower);
+ awaitConfigNodeStatus(location.getConfigNodeId(), NodeStatus.Running);
+ stopForcibly(EnvFactory.getEnv().getConfigNodeWrapper(follower));
+ awaitConfigNodeStatus(location.getConfigNodeId(), NodeStatus.Unknown);
}
@Test
- public void testForcedLeaderShutdownWithTwoDataNodes() throws Exception {
- init(2);
- int stopped = stoppedFixture(0);
- int leader = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- EnvFactory.getEnv().getConfigNodeWrapper(leader).stopForcibly();
- assertFalse(EnvFactory.getEnv().getConfigNodeWrapper(leader).isAlive());
- assertNotEquals(leader, EnvFactory.getEnv().getLeaderConfigNodeIndex());
- assertStatus(0, stopped, NodeStatus.Stopped);
- EnvFactory.getEnv()
- .ensureNodeStatus(
-
Collections.singletonList(EnvFactory.getEnv().getConfigNodeWrapper(leader)),
- Collections.singletonList(NodeStatus.Unknown));
- EnvFactory.getEnv().startConfigNode(leader);
- awaitConfigNodeRunning(leader);
- awaitPersisted(leader, stopped, NodeStatus.Stopped);
- transfer(leader);
- assertStatus(0, stopped, NodeStatus.Stopped);
- }
+ public void testStoppedDataNodeSurvivesLeaderFailure() throws Exception {
+ init(1);
+ int stopped = stopAndReportDataNode(0);
+ awaitDataNodeStatus(stopped, NodeStatus.Stopped, null);
+ int leader = awaitLeader();
+ stopForcibly(EnvFactory.getEnv().getConfigNodeWrapper(leader));
+ assertNotEquals(leader, awaitLeader());
+ // The DataNode stays offline, so fresh heartbeats cannot hide lost
persisted state.
+ awaitDataNodeStatus(stopped, NodeStatus.Stopped, null);
- @Test
- public void testNewLeaderChangesSurviveReturnToOriginalLeader() throws
Exception {
- init(3);
- int stopped = stoppedFixture(0);
- int removing = removingFixture(1);
- int first = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- int second = (first + 1) % 3;
- transfer(second);
- assertStatus(0, stopped, NodeStatus.Stopped);
- assertStatus(1, removing, NodeStatus.Removing);
- EnvFactory.getEnv().startDataNode(0);
- assertStatus(0, stopped, NodeStatus.Running);
- assertSuccess(control(second, "update:" + removing + ":Running"));
- awaitPersisted(first, stopped, null);
- awaitPersisted(first, removing, null);
- transfer((first + 2) % 3);
- transfer(first);
- EnvFactory.getEnv().getDataNodeWrapper(0).stopForcibly();
- assertStatus(0, stopped, NodeStatus.Unknown);
- transfer(second);
- assertStatus(0, stopped, NodeStatus.Unknown);
- assertStatus(1, removing, NodeStatus.Unknown);
+ EnvFactory.getEnv().startConfigNode(leader);
+ awaitConfigNodeStatus(configLocation(cluster(), leader).getConfigNodeId(),
NodeStatus.Running);
+ awaitDataNodeStatus(stopped, NodeStatus.Stopped, null);
}
@Test
- public void
testReadOnlyReasonPersistsAcrossLeaderTransferAndClearsAfterRestart()
- throws Exception {
- init(2);
+ public void testReadOnlyReasonSurvivesLeaderFailureAndClearsAfterRestart()
throws Exception {
+ init(1);
TDataNodeLocation dataNode = dataLocation(cluster(), 0);
- int dataNodeId = dataNode.getDataNodeId();
- IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> pool =
+ IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> clients =
new IClientManager.Factory<TEndPoint,
SyncDataNodeInternalServiceClient>()
.createClientManager(
new
ClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory());
try (SyncDataNodeInternalServiceClient client =
- pool.borrowClient(dataNode.getInternalEndPoint())) {
- assertEquals(
- TSStatusCode.SUCCESS_STATUS.getStatusCode(),
- client.setSystemStatus(NodeStatus.ReadOnly.getStatus()).getCode());
+ clients.borrowClient(dataNode.getInternalEndPoint())) {
+ assertSuccess(client.setSystemStatus(NodeStatus.ReadOnly.getStatus()));
} finally {
- pool.close();
- }
- assertStatus(0, dataNodeId, NodeStatus.ReadOnly);
- assertEquals(NodeStatus.MANUAL,
cluster().getNodeStatusReason().get(dataNodeId));
- // Inspect every voter's applied record before election; SHOW alone could
pass on heartbeats.
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, dataNodeId, NodeStatus.ReadOnly, NodeStatus.MANUAL);
- }
-
- int nextLeader = (EnvFactory.getEnv().getLeaderConfigNodeIndex() + 1) % 3;
- transfer(nextLeader);
- assertEquals(nextLeader, EnvFactory.getEnv().getLeaderConfigNodeIndex());
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, dataNodeId, NodeStatus.ReadOnly, NodeStatus.MANUAL);
- }
- assertStatus(0, dataNodeId, NodeStatus.ReadOnly);
- assertEquals(NodeStatus.MANUAL,
cluster().getNodeStatusReason().get(dataNodeId));
-
- // Explicitly report after stopping: Windows process termination does not
run shutdown hooks.
- assertEquals(dataNodeId, stoppedFixture(0));
- assertNull(cluster().getNodeStatusReason().get(dataNodeId));
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, dataNodeId, NodeStatus.Stopped, null);
+ clients.close();
}
-
+ int nodeId = dataNode.getDataNodeId();
+ awaitDataNodeStatus(nodeId, NodeStatus.ReadOnly, NodeStatus.MANUAL);
+ int leader = awaitLeader();
+ stopForcibly(EnvFactory.getEnv().getConfigNodeWrapper(leader));
+ assertNotEquals(leader, awaitLeader());
+ // This checks the visible status and reason; live heartbeats may rebuild
both.
+ awaitDataNodeStatus(nodeId, NodeStatus.ReadOnly, NodeStatus.MANUAL);
+
+ assertEquals(nodeId, stopAndReportDataNode(0));
+ awaitDataNodeStatus(nodeId, NodeStatus.Stopped, null);
EnvFactory.getEnv().startDataNode(0);
- assertStatus(0, dataNodeId, NodeStatus.Running);
- assertNull(cluster().getNodeStatusReason().get(dataNodeId));
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, dataNodeId, null, null);
- }
+ awaitDataNodeStatus(nodeId, NodeStatus.Running, null);
}
@Test
- public void testLiveOldLeaderPartitionAndQuorumRecovery() throws Exception {
- init(3);
- int stopped = stoppedFixture(0);
- int first = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- TConfigNodeLocation firstLocation = configLocation(cluster(), first);
- for (int i = 0; i < 3; i++) {
- control(i, "partition:" + firstLocation.getConfigNodeId());
- }
- int majorityLeader = awaitOtherLeader(first);
- assertTrue(
- "Partition must leave the old leader process alive",
- EnvFactory.getEnv().getConfigNodeWrapper(first).isAlive());
- assertSuccess(control(majorityLeader, "update:" + stopped + ":Removing"));
- try {
- assertStatus(0, stopped, NodeStatus.Removing);
- } catch (SQLException e) {
- // Preserve a failure for SQL routing without skipping consensus
recovery assertions.
- failures.addError(
- new AssertionError(
- "SHOW through the healthy DataNode cannot reach the elected
majority", e));
- }
- heal();
- awaitPersisted(first, stopped, NodeStatus.Removing);
- transfer(first);
- assertStatus(0, stopped, NodeStatus.Removing);
+ public void testStoppedStatusRecoversAndClearsAcrossConfigNodeRestarts()
throws Exception {
+ init(1);
+ int nodeId = stopAndReportDataNode(0);
+ awaitDataNodeStatus(nodeId, NodeStatus.Stopped, null);
+ restartConfigNodes();
+ awaitDataNodeStatus(nodeId, NodeStatus.Stopped, null);
- // Keep every process alive but prevent any pair of ConfigRegion voters
communicating.
- for (int i = 0; i < 3; i++) {
- control(i, "partition:*");
- }
- Properties noQuorum = control(first, "update:" + stopped + ":Running");
- assertNotEquals(
- "No quorum cannot confirm clearing Removing", "200",
noQuorum.getProperty("code"));
- assertEquals(
- "Removing", control(first, "inspect:" +
stopped).getProperty("persisted." + stopped));
- heal();
- int recovered = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- assertSuccess(control(recovered, "update:" + stopped + ":Running"));
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, stopped, null);
- }
- transfer((recovered + 1) % 3);
- assertStatus(0, stopped, NodeStatus.Unknown);
+ EnvFactory.getEnv().startDataNode(0);
+ awaitDataNodeStatus(nodeId, NodeStatus.Running, null);
+ // Do not report the second stop. The old Stopped marker must not return
after recovery.
+ stopForcibly(EnvFactory.getEnv().getDataNodeWrapper(0));
+ awaitDataNodeStatus(nodeId, NodeStatus.Unknown, null);
+ restartConfigNodes();
+ awaitDataNodeStatus(nodeId, NodeStatus.Unknown, null);
}
- private void init(int dataNodes) throws Exception {
+ @Test
+ public void testRemovedDataNodeStaysUnregisteredAfterLeaderFailure() throws
Exception {
+ // The production removal API does not support SimpleConsensus.
EnvFactory.getEnv()
.getConfig()
.getCommonConfig()
- .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
- .setConfigNodeRatisSnapshotTriggerThreshold(Integer.MAX_VALUE)
- .setConfigRegionRatisRPCLeaderElectionTimeoutMaxMs(4000);
- EnvFactory.getEnv().initClusterEnvironment(3, dataNodes);
- queryNode = dataNodes - 1;
- agentJar = temporary.getRoot().toPath().resolve("node-status-agent.jar");
- Manifest manifest = new Manifest();
- manifest.getMainAttributes().put(Attributes.Name.MANIFEST_VERSION, "1.0");
- manifest.getMainAttributes().putValue("Agent-Class",
NodeStatusTestAgent.class.getName());
- String resource = NodeStatusTestAgent.class.getName().replace('.', '/') +
".class";
- try (JarOutputStream jar = new
JarOutputStream(Files.newOutputStream(agentJar), manifest);
- InputStream bytecode =
getClass().getClassLoader().getResourceAsStream(resource)) {
- assertNotNull(bytecode);
- jar.putNextEntry(new JarEntry(resource));
- bytecode.transferTo(jar);
- jar.closeEntry();
- }
- }
-
- @Test
- public void testNoQuorumTransitionMatrix() throws Exception {
+
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setSchemaReplicationFactor(1)
+ .setDataReplicationFactor(1);
init(2);
- int nodeId = stoppedFixture(0);
- NodeStatus[][] transitions = {
- {NodeStatus.Running, NodeStatus.Stopped},
- {NodeStatus.Stopped, NodeStatus.Running},
- {NodeStatus.Removing, NodeStatus.Running},
- {NodeStatus.Removing, NodeStatus.Stopped}
- };
- for (NodeStatus[] transition : transitions) {
- int leader = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- assertSuccess(control(leader, "update:" + nodeId + ":" + transition[0]));
- NodeStatus persisted = transition[0] == NodeStatus.Running ? null :
transition[0];
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, nodeId, persisted);
- }
- try {
- for (int i = 0; i < 3; i++) {
- control(i, "partition:*");
- }
- Properties result = awaitFile(submit(leader, "update:" + nodeId + ":"
+ transition[1]));
- assertNotEquals(
- "No quorum: " + transition[0] + " -> " + transition[1] + " " +
result,
- "200",
- result.getProperty("code"));
- for (int i = 0; i < 3; i++) {
- assertEquals(
- String.valueOf(persisted),
- control(i, "inspect:" + nodeId).getProperty("persisted." +
nodeId));
- }
- // A read may reject or return a local cache before the old leader
steps down. Record
- // the actual direct RPC response rather than assert an undocumented
linearizable read.
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getConfigNodeConnection(leader)) {
- TShowClusterResp read = client.showCluster();
- System.out.println(
- "No-quorum SHOW RPC: " + transition[0] + " -> " + transition[1]
+ ", " + read);
- }
- try (Connection connection = queryConnection();
- Statement statement = connection.createStatement()) {
- try (ResultSet rows = statement.executeQuery("SHOW DATANODES")) {
- while (rows.next()) {
- if (rows.getInt(1) == nodeId) {
- System.out.println("No-quorum SHOW SQL cached status: " +
rows.getString("Status"));
- }
- }
- }
- } catch (SQLException unavailable) {
- System.out.println("No-quorum SHOW SQL unavailable: " +
unavailable.getMessage());
- }
- } finally {
- heal();
- }
- int recovered = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- assertSuccess(control(recovered, "update:" + nodeId + ":" +
transition[1]));
- NodeStatus finalPersisted = transition[1] == NodeStatus.Running ? null :
transition[1];
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, nodeId, finalPersisted);
- }
+ TShowClusterResp before = cluster();
+ TDataNodeLocation victim = dataLocation(before, 0);
+ int survivor = dataLocation(before, 1).getDataNodeId();
+ try (SyncConfigNodeIServiceClient client = leaderClient()) {
+ assertSuccess(
+ client
+ .removeDataNode(new
TDataNodeRemoveReq(Collections.singletonList(victim)))
+ .getStatus());
}
+ // Observe the completed procedure; Removing is transient and need not be
sampled.
+ awaitRemoved(victim.getDataNodeId(), survivor);
+ int leader = awaitLeader();
+ stopForcibly(EnvFactory.getEnv().getConfigNodeWrapper(leader));
+ assertNotEquals(leader, awaitLeader());
+ awaitRemoved(victim.getDataNodeId(), survivor);
}
- @Test
- public void testSnapshotAndFollowingClearLogRecoverTogether() throws
Exception {
- init(3);
- int stopped = stoppedFixture(0);
- int removing = removingFixture(1);
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, stopped, NodeStatus.Stopped);
- awaitPersisted(i, removing, NodeStatus.Removing);
- control(i, "snapshot");
- assertSnapshot(i, stopped, NodeStatus.Stopped, removing,
NodeStatus.Removing);
- }
- int leader = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- assertSuccess(control(leader, "update:" + stopped + ":Running"));
- assertSuccess(control(leader, "update:" + removing + ":Stopped"));
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, stopped, null);
- awaitPersisted(i, removing, NodeStatus.Stopped);
- }
- restartConfigNodes();
- assertStatus(0, stopped, NodeStatus.Unknown);
- assertStatus(1, removing, NodeStatus.Stopped);
- for (int i = 0; i < 3; i++) {
- control(i, "snapshot");
- assertSnapshot(i, stopped, null, removing, NodeStatus.Stopped);
- }
- restartConfigNodes();
- assertStatus(0, stopped, NodeStatus.Unknown);
- assertStatus(1, removing, NodeStatus.Stopped);
- }
-
- private void assertSnapshot(
- int configNode, int firstId, NodeStatus first, int secondId, NodeStatus
second)
- throws Exception {
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30);
- do {
- List<Path> snapshots;
- // Stop at sm/<term>_<index>, without opening temporary snapshot
directories on Windows.
- try (Stream<Path> paths =
- Files.walk(
- Paths.get(
-
EnvFactory.getEnv().getConfigNodeWrapper(configNode).getNodePath(),
- "data",
- "confignode",
- "consensus"),
- 3)) {
- snapshots =
- paths
- .filter(
- p ->
- p.getFileName().toString().matches("\\d+_\\d+")
- &&
p.getParent().getFileName().toString().equals("sm"))
- .collect(Collectors.toList());
- } catch (IOException | UncheckedIOException e) {
- // Ratis may delete an old snapshot while the directories are being
enumerated.
- TimeUnit.MILLISECONDS.sleep(200);
- continue;
- }
- for (Path snapshot : snapshots) {
- NodeInfo restored = new NodeInfo();
- try {
- restored.processLoadSnapshot(snapshot.toFile());
- } catch (IOException e) {
- // The snapshot may have been replaced since it was enumerated.
- continue;
- }
- if (java.util.Objects.equals(
- first == null ? null : new Pair<>(first, null),
- restored.getPersistedNodeStatus(firstId))
- && java.util.Objects.equals(
- second == null ? null : new Pair<>(second, null),
- restored.getPersistedNodeStatus(secondId))) {
- return;
- }
- }
- TimeUnit.MILLISECONDS.sleep(200);
- } while (System.nanoTime() < deadline);
- fail("No completed snapshot has the expected durable markers on ConfigNode
" + configNode);
- }
-
- private void restartConfigNodes() throws Exception {
- EnvFactory.getEnv().getConfigNodeWrapperList().forEach(n ->
n.stopForcibly());
- EnvFactory.getEnv().getConfigNodeWrapperList().forEach(n -> n.start());
- EnvFactory.getEnv().getLeaderConfigNodeIndex();
- EnvFactory.getEnv()
- .ensureNodeStatus(
- new
java.util.ArrayList<>(EnvFactory.getEnv().getConfigNodeWrapperList()),
- java.util.Arrays.asList(NodeStatus.Running, NodeStatus.Running,
NodeStatus.Running));
- }
-
- private void awaitConfigNodeRunning(int index) {
- EnvFactory.getEnv()
- .ensureNodeStatus(
-
Collections.singletonList(EnvFactory.getEnv().getConfigNodeWrapper(index)),
- Collections.singletonList(NodeStatus.Running));
- }
-
- @Test
- public void testRealRemovalPersistsIntentAndDeletesRegistration() throws
Exception {
+ private void init(int dataNodes) throws Exception {
EnvFactory.getEnv()
.getConfig()
.getCommonConfig()
-
.setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
- .setDataRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
- .setSchemaReplicationFactor(2)
- .setDataReplicationFactor(2);
- init(3);
- int leader = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- int victim = -1;
- int victimIndex = -1;
- try (Connection connection = queryConnection();
- Statement statement = connection.createStatement()) {
- statement.execute("CREATE DATABASE root.status_remove");
- statement.execute(
- "CREATE TIMESERIES root.status_remove.d.s WITH DATATYPE=INT32,
ENCODING=RLE");
- statement.execute("INSERT INTO root.status_remove.d(time,s) VALUES
(1,1)");
- TShowClusterResp before = cluster();
- int queryId = dataLocation(before, queryNode).getDataNodeId();
- try (ResultSet regions = statement.executeQuery("SHOW REGIONS")) {
- while (regions.next()) {
- int candidate = regions.getInt("DataNodeId");
- if (candidate != queryId) {
- victim = candidate;
- break;
- }
- }
- }
- assertTrue("The victim must host a real region replica", victim >= 0);
- for (int i = 0; i < 3; i++) {
- if (dataLocation(before, i).getDataNodeId() == victim) {
- victimIndex = i;
- }
- }
- assertTrue(victimIndex >= 0);
- Path barrier = submit(leader, "barrier");
- assertNull(awaitFile(barrier).getProperty("error"));
- try {
- statement.execute("REMOVE DATANODE " + victim);
- awaitFile(Paths.get(barrier + ".reached"));
- assertStatus(victimIndex, victim, NodeStatus.Removing);
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, victim, NodeStatus.Removing);
- }
- } finally {
- Files.createFile(Paths.get(barrier + ".release"));
- }
- long deadline = System.nanoTime() + TimeUnit.MINUTES.toNanos(3);
- boolean removed = false;
- do {
- final int victimId = victim;
- removed =
- cluster().getDataNodeList().stream().noneMatch(n ->
n.getDataNodeId() == victimId);
- if (!removed) {
- TimeUnit.MILLISECONDS.sleep(200);
- }
- } while (!removed && System.nanoTime() < deadline);
- assertTrue("RemoveDataNodesProcedure must unregister the node", removed);
- for (int i = 0; i < 3; i++) {
- awaitPersisted(i, victim, null);
- }
- transfer((leader + 1) % 3);
- try (ResultSet rows = statement.executeQuery("SHOW DATANODES")) {
- while (rows.next()) {
- assertNotEquals(victim, rows.getInt(1));
- }
- }
- try (ResultSet rows = statement.executeQuery("SELECT s FROM
root.status_remove.d")) {
- assertTrue(rows.next());
- assertEquals(1, rows.getInt(2));
- assertFalse(rows.next());
- }
- }
+ .setConfigNodeConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
+ .setConfigNodeRatisSnapshotTriggerThreshold(Integer.MAX_VALUE)
+ .setConfigRegionRatisRPCLeaderElectionTimeoutMaxMs(4000);
+ EnvFactory.getEnv().initClusterEnvironment(3, dataNodes);
+ configNodeClients =
+ new IClientManager.Factory<TEndPoint, SyncConfigNodeIServiceClient>()
+ .createClientManager(new
ClientPoolFactory.SyncConfigNodeIServiceClientPoolFactory());
+ // Fault-window observations use CN RPCs, which do not open a DataNode
login session.
+ // This suite does not inspect individual replicas or assert a particular
snapshot boundary.
}
- private int stoppedFixture(int index) throws Exception {
+ private int stopAndReportDataNode(int index) throws Exception {
TDataNodeLocation node = dataLocation(cluster(), index);
- EnvFactory.getEnv().getDataNodeWrapper(index).stopForcibly();
- assertFalse(EnvFactory.getEnv().getDataNodeWrapper(index).isAlive());
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- assertEquals(200, client.reportDataNodeShutdown(node).getCode());
+ stopForcibly(EnvFactory.getEnv().getDataNodeWrapper(index));
+ try (SyncConfigNodeIServiceClient client = leaderClient()) {
+ // Success acknowledges the persisted report, unlike observing a
heartbeat-derived status.
+ assertSuccess(client.reportDataNodeShutdown(node));
}
- assertStatus(index, node.getDataNodeId(), NodeStatus.Stopped);
return node.getDataNodeId();
}
- private int removingFixture(int index) throws Exception {
- TDataNodeLocation node = dataLocation(cluster(), index);
- IClientManager<TEndPoint, SyncDataNodeInternalServiceClient> pool =
- new IClientManager.Factory<TEndPoint,
SyncDataNodeInternalServiceClient>()
- .createClientManager(
- new
ClientPoolFactory.SyncDataNodeInternalServiceClientPoolFactory());
- try (SyncDataNodeInternalServiceClient client =
pool.borrowClient(node.getInternalEndPoint())) {
- assertEquals(200, client.setSystemStatus("Removing").getCode());
- } finally {
- pool.close();
- }
- assertStatus(index, node.getDataNodeId(), NodeStatus.Removing);
- EnvFactory.getEnv().getDataNodeWrapper(index).stopForcibly();
- return node.getDataNodeId();
+ private void stopForcibly(AbstractNodeWrapper node) throws
InterruptedException {
+ node.stopForcibly();
+ assertTrue(node.getId(), node.getInstance().waitFor(60, TimeUnit.SECONDS));
+ assertFalse(node.getId(), node.isAlive());
}
- private void assertStatus(int index, int nodeId, NodeStatus expected) throws
Exception {
- EnvFactory.getEnv()
- .ensureNodeStatus(
-
Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(index)),
- Collections.singletonList(expected));
- assertEquals(expected.getStatus(), cluster().getNodeStatus().get(nodeId));
- // Pin the SQL connection to the healthy query node; do not fan out to
stopped DataNodes.
- try (Connection connection = queryConnection();
- Statement statement = connection.createStatement()) {
- for (String sql : new String[] {"SHOW CLUSTER", "SHOW DATANODES"}) {
- boolean found = false;
- try (ResultSet rows = statement.executeQuery(sql)) {
- while (rows.next()) {
- if (rows.getInt(1) == nodeId) {
- assertEquals(sql, expected.getStatus(),
rows.getString("Status"));
- found = true;
- }
- }
- }
- assertTrue(sql + " did not contain DataNode " + nodeId, found);
- }
+ private void restartConfigNodes() throws Exception {
+ List<ConfigNodeWrapper> nodes =
EnvFactory.getEnv().getConfigNodeWrapperList();
+ for (ConfigNodeWrapper node : nodes) {
+ stopForcibly(node);
}
+ configNodeClients.clearAll();
+ nodes.forEach(ConfigNodeWrapper::start);
+ awaitCluster(
+ "All ConfigNodes must recover",
+ (client, cluster) -> {
+ assertEquals(3, cluster.getConfigNodeListSize());
+ for (TConfigNodeLocation node : cluster.getConfigNodeList()) {
+ assertEquals(
+ cluster.toString(),
+ NodeStatus.Running.getStatus(),
+ cluster.getNodeStatus().get(node.getConfigNodeId()));
+ }
+ });
}
- private void transfer(int target) throws Exception {
- int leader = EnvFactory.getEnv().getLeaderConfigNodeIndex();
- if (leader == target) {
- return;
- }
- long previousTerm = Long.parseLong(control(leader,
"inspect").getProperty("term"));
- TConfigNodeLocation location = configLocation(cluster(), target);
- control(
- leader,
- "transfer:" + location.getConfigNodeId() + ":" +
location.getConsensusEndPoint().getPort());
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
- do {
- if (EnvFactory.getEnv().getLeaderConfigNodeIndex() == target) {
- assertTrue(Long.parseLong(control(target,
"inspect").getProperty("term")) > previousTerm);
- return;
- }
- TimeUnit.MILLISECONDS.sleep(200);
- } while (System.nanoTime() < deadline);
- fail("Leadership did not transfer to ConfigNode " + target);
+ private void awaitDataNodeStatus(int nodeId, NodeStatus status, String
reason)
+ throws InterruptedException {
+ awaitCluster(
+ "DataNode " + nodeId + " must become " + status + " with reason " +
reason,
+ (client, cluster) -> {
+ assertTrue(
+ cluster.toString(),
+ cluster.getDataNodeList().stream().anyMatch(n ->
n.getDataNodeId() == nodeId));
+ assertEquals(cluster.toString(), status.getStatus(),
cluster.getNodeStatus().get(nodeId));
+ assertEquals(cluster.toString(), reason,
cluster.getNodeStatusReason().get(nodeId));
+ TShowDataNodesResp response = client.showDataNodes();
+ assertSuccess(response.getStatus());
+ TDataNodeInfo node =
+ response.getDataNodesInfoList().stream()
+ .filter(n -> n.getDataNodeId() == nodeId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError(response));
+ assertEquals(response.toString(), status.getStatus(),
node.getStatus());
+ assertEquals(response.toString(), reason, node.getStatusReason());
+ });
}
- private int awaitOtherLeader(int excluded) throws Exception {
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
- do {
- for (int i = 0; i < 3; i++) {
- if (i != excluded) {
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getConfigNodeConnection(i)) {
- if (client.showCluster().getStatus().getCode() == 200) {
- return i;
- }
- } catch (Exception ignored) {
- /* Election can temporarily reject a read. */
- }
- }
- }
- TimeUnit.MILLISECONDS.sleep(200);
- } while (System.nanoTime() < deadline);
- throw new AssertionError("Majority did not elect a ready leader");
+ private void awaitConfigNodeStatus(int nodeId, NodeStatus status) throws
InterruptedException {
+ awaitCluster(
+ "ConfigNode " + nodeId + " must become " + status,
+ (client, cluster) -> {
+ assertTrue(
+ cluster.toString(),
+ cluster.getConfigNodeList().stream().anyMatch(n ->
n.getConfigNodeId() == nodeId));
+ assertEquals(cluster.toString(), status.getStatus(),
cluster.getNodeStatus().get(nodeId));
+ TShowConfigNodesResp response = client.showConfigNodes();
+ assertSuccess(response.getStatus());
+ TConfigNodeInfo node =
+ response.getConfigNodesInfoList().stream()
+ .filter(n -> n.getConfigNodeId() == nodeId)
+ .findFirst()
+ .orElseThrow(() -> new AssertionError(response));
+ assertEquals(response.toString(), status.getStatus(),
node.getStatus());
+ });
}
- private void heal() throws Exception {
- for (int i = 0; i < 3; i++) {
- control(i, "heal");
- }
+ private void awaitRemoved(int removed, int remaining) throws
InterruptedException {
+ awaitCluster(
+ "DataNode " + removed + " must stay unregistered",
+ (client, cluster) -> {
+ assertEquals(
+ cluster.toString(),
+ Collections.singletonList(remaining),
+ cluster.getDataNodeList().stream()
+ .map(TDataNodeLocation::getDataNodeId)
+ .collect(Collectors.toList()));
+ assertFalse(cluster.toString(),
cluster.getNodeStatus().containsKey(removed));
+ TShowDataNodesResp response = client.showDataNodes();
+ assertSuccess(response.getStatus());
+ assertEquals(
+ response.toString(),
+ Collections.singletonList(remaining),
+ response.getDataNodesInfoList().stream()
+ .map(TDataNodeInfo::getDataNodeId)
+ .collect(Collectors.toList()));
+ });
}
- private void awaitPersisted(int node, int dataNode, NodeStatus expected)
throws Exception {
- awaitPersisted(node, dataNode, expected, null);
+ private int awaitLeader() throws InterruptedException {
+ return awaitCluster("A ConfigNode leader must be ready", (client, cluster)
-> {});
}
- private void awaitPersisted(int node, int dataNode, NodeStatus expected,
String expectedReason)
- throws Exception {
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60);
+ private int awaitCluster(String description, ClusterAssertion assertion)
+ throws InterruptedException {
+ long deadline = System.nanoTime() + TimeUnit.MINUTES.toNanos(2);
+ Throwable lastFailure = null;
+ List<ConfigNodeWrapper> nodes =
EnvFactory.getEnv().getConfigNodeWrapperList();
do {
- Properties state = control(node, "inspect:" + dataNode);
- if (String.valueOf(expected).equals(state.getProperty("persisted." +
dataNode))
- && String.valueOf(expectedReason)
- .equals(state.getProperty("persistedReason." + dataNode))) {
- return;
+ for (int i = 0; i < nodes.size(); i++) {
+ if (!nodes.get(i).isAlive()) {
+ continue;
+ }
+ try (SyncConfigNodeIServiceClient client = configNodeClient(i)) {
+ TShowClusterResp response = client.showCluster();
+ assertSuccess(response.getStatus());
+ // A successful showCluster also checks consensus and leader-service
readiness.
+ assertion.check(client, response);
+ return i;
+ } catch (ClientManagerException | TException | AssertionError e) {
+ lastFailure = e;
+ }
}
- TimeUnit.MILLISECONDS.sleep(200);
+ TimeUnit.SECONDS.sleep(1);
} while (System.nanoTime() < deadline);
- fail(
- "ConfigNode "
- + node
- + " did not persist "
- + expected
- + " with reason "
- + expectedReason
- + " for "
- + dataNode);
+ throw new AssertionError(description, lastFailure);
}
- private Path submit(int node, String command) throws Exception {
- Path output =
- temporary.getRoot().toPath().resolve("control-" + System.nanoTime() +
".properties");
- VirtualMachine vm =
- VirtualMachine.attach(
-
Long.toString(EnvFactory.getEnv().getConfigNodeWrapper(node).getInstance().pid()));
- try {
- vm.loadAgent(agentJar.toString(), command + "|" + output);
- } finally {
- vm.detach();
- }
- return output;
+ private SyncConfigNodeIServiceClient configNodeClient(int index) throws
ClientManagerException {
+ ConfigNodeWrapper node = EnvFactory.getEnv().getConfigNodeWrapper(index);
+ return configNodeClients.borrowClient(new TEndPoint(node.getIp(),
node.getPort()));
}
- private Properties control(int node, String command) throws Exception {
- Properties result = awaitFile(submit(node, command));
- assertNull(command + " failed: " + result, result.getProperty("error"));
- return result;
- }
-
- private Properties awaitFile(Path output) throws Exception {
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(90);
- while (!Files.exists(output) && System.nanoTime() < deadline) {
- TimeUnit.MILLISECONDS.sleep(100);
- }
- assertTrue("Test control timed out: " + output, Files.exists(output));
- Properties result = new Properties();
- try (InputStream stream = Files.newInputStream(output)) {
- result.load(stream);
- }
- return result;
+ private SyncConfigNodeIServiceClient leaderClient()
+ throws InterruptedException, ClientManagerException {
+ return configNodeClient(awaitLeader());
}
private TShowClusterResp cluster() throws Exception {
- try (SyncConfigNodeIServiceClient client =
- (SyncConfigNodeIServiceClient)
EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
- TShowClusterResp result = client.showCluster();
- assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(),
result.getStatus().getCode());
- return result;
+ try (SyncConfigNodeIServiceClient client = leaderClient()) {
+ TShowClusterResp response = client.showCluster();
+ assertSuccess(response.getStatus());
+ return response;
}
}
@@ -722,22 +378,12 @@ public class IoTDBNodeStatusFailoverIT {
.orElseThrow(AssertionError::new);
}
- private void assertSuccess(Properties result) {
- assertEquals(result.toString(), "200", result.getProperty("code"));
+ private void assertSuccess(TSStatus status) {
+ assertEquals(status.toString(),
TSStatusCode.SUCCESS_STATUS.getStatusCode(), status.getCode());
}
- private Connection queryConnection() throws SQLException {
- Properties properties = new Properties();
- properties.setProperty("user", "root");
- properties.setProperty("password", "root");
- properties.setProperty("network_timeout", "15000");
- properties.setProperty("sql_dialect", "tree");
- return DriverManager.getConnection(
- "jdbc:iotdb://"
- + EnvFactory.getEnv().getDataNodeWrapper(queryNode).getIp()
- + ":"
- + EnvFactory.getEnv().getDataNodeWrapper(queryNode).getPort()
- + "/",
- properties);
+ @FunctionalInterface
+ private interface ClusterAssertion {
+ void check(SyncConfigNodeIServiceClient client, TShowClusterResp cluster)
throws TException;
}
}
diff --git
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/NodeStatusTestAgent.java
b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/NodeStatusTestAgent.java
deleted file mode 100644
index 529afd9bccc..00000000000
---
a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/NodeStatusTestAgent.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/*
- * 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.it.cluster;
-
-import org.apache.iotdb.common.rpc.thrift.TEndPoint;
-import org.apache.iotdb.common.rpc.thrift.TSStatus;
-import org.apache.iotdb.commons.client.request.AsyncRequestManager;
-import org.apache.iotdb.commons.cluster.NodeStatus;
-import org.apache.iotdb.commons.utils.function.CheckedTriConsumer;
-import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
-import
org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
-import org.apache.iotdb.confignode.manager.ConfigManager;
-import org.apache.iotdb.confignode.manager.consensus.ConsensusManager;
-import org.apache.iotdb.confignode.persistence.node.NodeInfo;
-import org.apache.iotdb.confignode.service.ConfigNode;
-import org.apache.iotdb.consensus.common.Peer;
-import org.apache.iotdb.consensus.ratis.utils.Utils;
-
-import com.google.common.collect.ImmutableMap;
-import org.apache.ratis.util.CodeInjectionForTesting;
-import org.apache.thrift.TException;
-import org.apache.tsfile.utils.Pair;
-
-import java.io.OutputStream;
-import java.lang.instrument.Instrumentation;
-import java.lang.reflect.Field;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.TimeUnit;
-
-/** Loaded only into test JVMs. No production endpoint or production bytecode
is changed. */
-public final class NodeStatusTestAgent {
- private NodeStatusTestAgent() {}
-
- public static void agentmain(String arguments, Instrumentation ignored) {
- String[] parts = arguments.split("\\|", 2);
- String command = parts[0];
- Path output = Paths.get(parts[1]);
- // Return from attach immediately even if a consensus write needs to time
out.
- Thread worker = new Thread(() -> execute(command, output),
"node-status-test-control");
- worker.setDaemon(true);
- worker.start();
- }
-
- private static void execute(String command, Path output) {
- Properties result = new Properties();
- try {
- String[] args = command.split(":");
- if ("shutdown".equals(args[0])) {
- Runtime.getRuntime().addShutdownHook(new Thread(() -> store(output,
result)));
- System.exit(0);
- return;
- }
- ConfigManager manager = ConfigNode.getInstance().getConfigManager();
- ConsensusManager consensus = manager.getConsensusManager();
- switch (args[0]) {
- case "snapshot":
-
consensus.getConsensusImpl().triggerSnapshot(consensus.getConsensusGroupId(),
true);
- break;
- case "barrier":
- installRemovalBarrier(output);
- break;
- case "partition":
- String isolated =
- "*".equals(args[1])
- ? "*"
- :
Utils.fromNodeIdToRaftPeerId(Integer.parseInt(args[1])).toString();
- for (String point :
- new String[] {"requestVote", "appendEntries",
"startLeaderElection"}) {
- CodeInjectionForTesting.put(
- "RaftServerImpl." + point,
- (local, remote, values) -> {
- if ("*".equals(isolated)
- || isolated.equals(String.valueOf(local))
- || isolated.equals(String.valueOf(remote))) {
- // Fail at receipt, before term/log processing. This drops
Raft messages in both
- // directions while leaving the JVM and ConfigNode
business RPC server alive.
- throw new IllegalStateException("Injected Raft message
partition");
- }
- return false;
- });
- }
- break;
- case "heal":
- for (String point :
- new String[] {"requestVote", "appendEntries",
"startLeaderElection"}) {
- CodeInjectionForTesting.remove("RaftServerImpl." + point);
- }
- break;
- case "transfer":
- consensus
- .getConsensusImpl()
- .transferLeader(
- consensus.getConsensusGroupId(),
- new Peer(
- consensus.getConsensusGroupId(),
- Integer.parseInt(args[1]),
- new TEndPoint("127.0.0.1", Integer.parseInt(args[2]))));
- break;
- case "update":
- int id = Integer.parseInt(args[1]);
- NodeStatus status = NodeStatus.parse(args[2]);
- TSStatus response = manager.getLoadManager().trySetNodeStatus(id,
status, true);
- result.setProperty("code", Integer.toString(response.getCode()));
- break;
- case "inspect":
- NodeInfo nodeInfo = manager.getNodeManager().getNodeInfo();
- for (int i = 1; i < args.length; i++) {
- int nodeId = Integer.parseInt(args[i]);
- Pair<NodeStatus, String> persistedStatus =
nodeInfo.getPersistedNodeStatus(nodeId);
- result.setProperty(
- "persisted." + nodeId,
- String.valueOf(persistedStatus == null ? null :
persistedStatus.left));
- result.setProperty(
- "persistedReason." + nodeId,
- String.valueOf(persistedStatus == null ? null :
persistedStatus.right));
- }
- break;
- default:
- throw new IllegalArgumentException(command);
- }
- result.setProperty(
- "term",
- Long.toString(
-
consensus.getConsensusImpl().getLogicalClock(consensus.getConsensusGroupId())));
- result.setProperty("leader", Boolean.toString(consensus.isLeader()));
- } catch (Throwable e) {
- result.setProperty("error", e.toString());
- }
- store(output, result);
- }
-
- private static void store(Path output, Properties result) {
- try {
- Path temporary = output.resolveSibling(output.getFileName() + ".tmp");
- try (OutputStream stream = Files.newOutputStream(temporary)) {
- result.store(stream, "Node status test control");
- }
- Files.move(temporary, output);
- } catch (Exception e) {
- throw new IllegalStateException(e);
- }
- }
-
- @SuppressWarnings({"rawtypes", "unchecked"})
- private static void installRemovalBarrier(Path output) throws Exception {
- CnToDnInternalServiceAsyncRequestManager requests =
- CnToDnInternalServiceAsyncRequestManager.getInstance();
- Field field = AsyncRequestManager.class.getDeclaredField("actionMap");
- field.setAccessible(true);
- Map actions = new HashMap((Map) field.get(requests));
- CheckedTriConsumer<Object, Object, Object, TException> original =
- (CheckedTriConsumer<Object, Object, Object, TException>)
- actions.get(CnToDnAsyncRequestType.CLEAN_DATA_NODE_CACHE);
- Path reached = Paths.get(output + ".reached");
- Path release = Paths.get(output + ".release");
- CheckedTriConsumer<Object, Object, Object, TException> gated =
- (request, client, callback) -> {
- if (!Files.exists(reached)) {
- store(reached, new Properties());
- }
- long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(120);
- while (!Files.exists(release)) {
- if (System.nanoTime() >= deadline) {
- throw new TException("Removal test barrier timed out");
- }
- try {
- TimeUnit.MILLISECONDS.sleep(100);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new TException(e);
- }
- }
- try {
- original.accept(request, client, callback);
- } catch (Exception e) {
- throw new TException(e);
- }
- };
- actions.put(CnToDnAsyncRequestType.CLEAN_DATA_NODE_CACHE, gated);
- field.set(requests, ImmutableMap.copyOf(actions));
- }
-}
