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 13dd39c7a3189..b968059759617 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 @@ 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 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 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 pool = + IClientManager clients = new IClientManager.Factory() .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 snapshots; - // Stop at sm/_, without opening temporary snapshot directories on Windows. - try (Stream 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() + .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 pool = - new IClientManager.Factory() - .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 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 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 @@ private TConfigNodeLocation configLocation(TShowClusterResp cluster, int index) .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 529afd9bccc85..0000000000000 --- 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 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 original = - (CheckedTriConsumer) - actions.get(CnToDnAsyncRequestType.CLEAN_DATA_NODE_CACHE); - Path reached = Paths.get(output + ".reached"); - Path release = Paths.get(output + ".release"); - CheckedTriConsumer 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)); - } -}