@Test public void testChangeNode() throws Exception { List<Broker> allNodes = new LinkedList<>(); for (int i = 0; i < NODE_NUM; i++) { allNodes.add(brokers[i]); } for (int i = 0; i < RAFT_ELECTION_NUM; i++) { storeServices[i].createPartitionGroup(topic1.getFullName(), partitionGroup1, partitions); electionManager[i].onPartitionGroupCreate(PartitionGroup.ElectType.raft, topic1, partitionGroup1, allNodes, new TreeSet<>(), brokers[i].getId(), -1); leaderElections[i] = electionManager[i].getLeaderElection(topic1, partitionGroup1); } int leaderId = getLeader(leaderElections[0], 10); Assert.assertNotEquals(leaderId, -1); logger.info("Leader id is " + leaderId); Assert.assertEquals(leaderId, leaderElections[1].getLeaderId()); Assert.assertEquals(leaderId, leaderElections[2].getLeaderId()); logger.info("Remove leader node {}", leaderId); for (int i = 0; i < RAFT_ELECTION_NUM; i++) { if (leaderId == i + 1) { electionManager[i].onPartitionGroupRemove(topic1, partitionGroup1); } else { electionManager[i].onNodeRemove(topic1, partitionGroup1, leaderId, brokers[i].getId()); } } Thread.sleep(10000); int leaderId1 = getLeader(leaderElections[nextNode(leaderId) - 1], 10); logger.info("Leader1 id is " + leaderId1); Assert.assertNotEquals(leaderId1, -1); Collection<DefaultElectionNode> allNodes1 = leaderElections[leaderId1 - 1].getAllNodes(); Assert.assertEquals(allNodes1.size(), 2); logger.info("Remove leader node {}", leaderId1); for (int i = 0; i < RAFT_ELECTION_NUM; i++) { if (i + 1 == leaderId) continue; if (leaderId1 == i + 1) { electionManager[i].onPartitionGroupRemove(topic1, partitionGroup1); } else { electionManager[i].onNodeRemove(topic1, partitionGroup1, leaderId1, brokers[i].getId()); } } Thread.sleep(5000); int i; for (i = 0; i < RAFT_ELECTION_NUM; i++) { if (leaderId != i + 1 && leaderId1 != i + 1) break; } logger.info("Remain node is {}", i); int leaderId2 = getLeader(leaderElections[i], 10); logger.info("Leader2 id is " + leaderId2); Assert.assertNotEquals(leaderId2, -1); Collection<DefaultElectionNode> allNodes2 = leaderElections[leaderId2 - 1].getAllNodes(); Assert.assertEquals(allNodes2.size(), 1); List<Broker> allNodesAdd = new LinkedList<>(); allNodesAdd.add(brokers[leaderId2 - 1]); allNodesAdd.add(brokers[leaderId1 - 1]); logger.info("Add node {}", leaderId1); electionManager[leaderId2 - 1].onNodeAdd(topic1, partitionGroup1, PartitionGroup.ElectType.raft, allNodesAdd, new TreeSet<>(), brokers[leaderId1 - 1], leaderId1, -1); electionManager[leaderId1 - 1].onPartitionGroupCreate(PartitionGroup.ElectType.raft, topic1, partitionGroup1, allNodesAdd, new TreeSet<>(), brokers[leaderId1 - 1].getId(), -1); leaderElections[leaderId1 - 1] = electionManager[i].getLeaderElection(topic1, partitionGroup1); int leaderId3 = getLeader(leaderElections[leaderId2 - 1], 10); logger.info("Leader3 id is " + leaderId3); Assert.assertNotEquals(leaderId3, -1); Collection<DefaultElectionNode> allNodes3 = leaderElections[leaderId2 - 1].getAllNodes(); Assert.assertEquals(allNodes3.size(), 2); Collection<DefaultElectionNode> allNodes4 = leaderElections[leaderId1 - 1].getAllNodes(); Assert.assertEquals(allNodes4.size(), 2); Thread.sleep(1000); }
@Override public Collection<DefaultElectionNode> getAllNodes() { return allNodes.values(); }
RaftLeaderElection extends LeaderElection { @Override public Collection<DefaultElectionNode> getAllNodes() { return allNodes.values(); } }
RaftLeaderElection extends LeaderElection { @Override public Collection<DefaultElectionNode> getAllNodes() { return allNodes.values(); } RaftLeaderElection(TopicPartitionGroup topicPartitionGroup, ElectionConfig electionConfig,
                       ElectionManager electionManager, ClusterManager clusterManager,
                       ElectionMetadataManager metadataManager, ReplicableStore replicableStore,
                       ReplicaGroup replicaGroup, ScheduledExecutorService electionTimerExecutor,
                       ExecutorService electionExecutor, EventBus<ElectionEvent> electionEventManager,
                       int localNodeId, List<DefaultElectionNode> allNodes,
                       Set<Integer> learners); }
RaftLeaderElection extends LeaderElection { @Override public Collection<DefaultElectionNode> getAllNodes() { return allNodes.values(); } RaftLeaderElection(TopicPartitionGroup topicPartitionGroup, ElectionConfig electionConfig,
                       ElectionManager electionManager, ClusterManager clusterManager,
                       ElectionMetadataManager metadataManager, ReplicableStore replicableStore,
                       ReplicaGroup replicaGroup, ScheduledExecutorService electionTimerExecutor,
                       ExecutorService electionExecutor, EventBus<ElectionEvent> electionEventManager,
                       int localNodeId, List<DefaultElectionNode> allNodes,
                       Set<Integer> learners); @Override Collection<DefaultElectionNode> getAllNodes(); ElectionNode.State state(); void state(ElectionNode.State state); @Override void addNode(DefaultElectionNode node); @Override void removeNode(int nodeId); @Override void setLeaderId(int leaderId); synchronized Command handleVoteRequest(VoteRequest voteRequest); synchronized Command handlePreVoteRequest(VoteRequest voteRequest); @Override synchronized Command handleAppendEntriesRequest(AppendEntriesRequest request); @Override synchronized void stepDown(int term); synchronized Command handleTimeoutNowRequest(TimeoutNowRequest request); }
RaftLeaderElection extends LeaderElection { @Override public Collection<DefaultElectionNode> getAllNodes() { return allNodes.values(); } RaftLeaderElection(TopicPartitionGroup topicPartitionGroup, ElectionConfig electionConfig,
                       ElectionManager electionManager, ClusterManager clusterManager,
                       ElectionMetadataManager metadataManager, ReplicableStore replicableStore,
                       ReplicaGroup replicaGroup, ScheduledExecutorService electionTimerExecutor,
                       ExecutorService electionExecutor, EventBus<ElectionEvent> electionEventManager,
                       int localNodeId, List<DefaultElectionNode> allNodes,
                       Set<Integer> learners); @Override Collection<DefaultElectionNode> getAllNodes(); ElectionNode.State state(); void state(ElectionNode.State state); @Override void addNode(DefaultElectionNode node); @Override void removeNode(int nodeId); @Override void setLeaderId(int leaderId); synchronized Command handleVoteRequest(VoteRequest voteRequest); synchronized Command handlePreVoteRequest(VoteRequest voteRequest); @Override synchronized Command handleAppendEntriesRequest(AppendEntriesRequest request); @Override synchronized void stepDown(int term); synchronized Command handleTimeoutNowRequest(TimeoutNowRequest request);  }