Candidate extends AbstractRaftActorBehavior { @Override public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) { if (message instanceof ElectionTimeout) { log.debug("{}: Received ElectionTimeout", logName()); if (votesRequired == 0) { return internalSwitchBehavior(RaftState.Leader); } startNewTerm(); scheduleElection(electionDuration()); return this; } if (message instanceof RaftRPC) { RaftRPC rpc = (RaftRPC) message; log.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc, context.getTermInformation().getCurrentTerm()); if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) { log.info("{}: Term {} in \"{}\" message is greater than Candidate's term {} - switching to Follower", logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm()); context.getTermInformation().updateAndPersist(rpc.getTerm(), null); if (message instanceof RequestVote) { super.handleMessage(sender, message); } return internalSwitchBehavior(RaftState.Follower); } } return super.handleMessage(sender, message); } Candidate(final RaftActorContext context); @Override final String getLeaderId(); @Override final short getLeaderPayloadVersion(); @Override RaftActorBehavior handleMessage(final ActorRef sender, final Object message); @Override void close();  }
@Test public void testHandleElectionTimeoutWhenThereAreZeroPeers() { RaftActorContext raftActorContext = createActorContext(); candidate = new Candidate(raftActorContext); RaftActorBehavior newBehavior = candidate.handleMessage(candidateActor, ElectionTimeout.INSTANCE); assertEquals("Behavior", RaftState.Leader, newBehavior.state()); }
@Test public void testHandleElectionTimeoutWhenThereAreTwoNodeCluster() { MockRaftActorContext raftActorContext = createActorContext(); raftActorContext.setPeerAddresses(setupPeers(1)); candidate = new Candidate(raftActorContext); candidate = candidate.handleMessage(candidateActor, ElectionTimeout.INSTANCE); assertEquals("Behavior", RaftState.Candidate, candidate.state()); }
@Test public void testBecomeLeaderOnReceivingMajorityVotesInThreeNodeCluster() { MockRaftActorContext raftActorContext = createActorContext(); raftActorContext.setLastApplied(raftActorContext.getReplicatedLog().lastIndex()); raftActorContext.setPeerAddresses(setupPeers(2)); candidate = new Candidate(raftActorContext); candidate = candidate.handleMessage(peerActors[0], new RequestVoteReply(1, true)); assertEquals("Behavior", RaftState.Leader, candidate.state()); }
@Test public void testBecomePreLeaderOnReceivingMajorityVotesInThreeNodeCluster() { MockRaftActorContext raftActorContext = createActorContext(); raftActorContext.setLastApplied(-1); raftActorContext.setPeerAddresses(setupPeers(2)); candidate = new Candidate(raftActorContext); candidate = candidate.handleMessage(peerActors[0], new RequestVoteReply(1, true)); assertEquals("Behavior", RaftState.PreLeader, candidate.state()); }
@Test public void testBecomeLeaderOnReceivingMajorityVotesInFiveNodeCluster() { MockRaftActorContext raftActorContext = createActorContext(); raftActorContext.getTermInformation().update(2L, "other"); raftActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder() .createEntries(0, 5, 1).build()); raftActorContext.setCommitIndex(raftActorContext.getReplicatedLog().lastIndex()); raftActorContext.setLastApplied(raftActorContext.getReplicatedLog().lastIndex()); raftActorContext.setPeerAddresses(setupPeers(4)); candidate = new Candidate(raftActorContext); RequestVote requestVote = MessageCollectorActor.expectFirstMatching(peerActors[0], RequestVote.class); assertEquals("getTerm", 3L, requestVote.getTerm()); assertEquals("getCandidateId", raftActorContext.getId(), requestVote.getCandidateId()); assertEquals("getLastLogTerm", 1L, requestVote.getLastLogTerm()); assertEquals("getLastLogIndex", 4L, requestVote.getLastLogIndex()); MessageCollectorActor.expectFirstMatching(peerActors[1], RequestVote.class); MessageCollectorActor.expectFirstMatching(peerActors[2], RequestVote.class); MessageCollectorActor.expectFirstMatching(peerActors[3], RequestVote.class); candidate = candidate.handleMessage(peerActors[0], new RequestVoteReply(1, false)); assertEquals("Behavior", RaftState.Candidate, candidate.state()); candidate = candidate.handleMessage(peerActors[1], new RequestVoteReply(1, true)); assertEquals("Behavior", RaftState.Candidate, candidate.state()); candidate = candidate.handleMessage(peerActors[2], new RequestVoteReply(1, true)); assertEquals("Behavior", RaftState.Leader, candidate.state()); }
@Test public void testBecomeLeaderOnReceivingMajorityVotesWithNonVotingPeers() { ElectionTerm mockElectionTerm = Mockito.mock(ElectionTerm.class); Mockito.doReturn(1L).when(mockElectionTerm).getCurrentTerm(); RaftActorContext raftActorContext = new RaftActorContextImpl(candidateActor, candidateActor.actorContext(), "candidate", mockElectionTerm, -1, -1, setupPeers(4), new DefaultConfigParamsImpl(), new NonPersistentDataProvider(Runnable::run), applyState -> { }, LOG, MoreExecutors.directExecutor()); raftActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); raftActorContext.getPeerInfo("peer1").setVotingState(VotingState.NON_VOTING); raftActorContext.getPeerInfo("peer4").setVotingState(VotingState.NON_VOTING); candidate = new Candidate(raftActorContext); MessageCollectorActor.expectFirstMatching(peerActors[1], RequestVote.class); MessageCollectorActor.expectFirstMatching(peerActors[2], RequestVote.class); MessageCollectorActor.assertNoneMatching(peerActors[0], RequestVote.class, 300); MessageCollectorActor.assertNoneMatching(peerActors[3], RequestVote.class, 100); candidate = candidate.handleMessage(peerActors[1], new RequestVoteReply(1, false)); assertEquals("Behavior", RaftState.Candidate, candidate.state()); candidate = candidate.handleMessage(peerActors[2], new RequestVoteReply(1, true)); assertEquals("Behavior", RaftState.Leader, candidate.state()); }
@Test public void testResponseToHandleAppendEntriesWithLowerTerm() { candidate = new Candidate(createActorContext()); setupPeers(1); RaftActorBehavior newBehavior = candidate.handleMessage(peerActors[0], new AppendEntries(1, "test", 0, 0, Collections.<ReplicatedLogEntry>emptyList(), 0, -1, (short) 0)); AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching( peerActors[0], AppendEntriesReply.class); assertEquals("isSuccess", false, reply.isSuccess()); assertEquals("getTerm", 2, reply.getTerm()); assertTrue("New Behavior : " + newBehavior, newBehavior instanceof Candidate); }
@Test public void testResponseToHandleAppendEntriesWithHigherTerm() { candidate = new Candidate(createActorContext()); setupPeers(1); RaftActorBehavior newBehavior = candidate.handleMessage(peerActors[0], new AppendEntries(5, "test", 0, 0, Collections.<ReplicatedLogEntry>emptyList(), 0, -1, (short) 0)); assertTrue("New Behavior : " + newBehavior, newBehavior instanceof Follower); }
@Test public void testResponseToHandleAppendEntriesWithEqualTerm() { MockRaftActorContext actorContext = createActorContext(); candidate = new Candidate(actorContext); setupPeers(1); RaftActorBehavior newBehavior = candidate.handleMessage(peerActors[0], new AppendEntries(2, "test", 0, 0, Collections.<ReplicatedLogEntry>emptyList(), 0, -1, (short) 0)); assertTrue("New Behavior : " + newBehavior + " term = " + actorContext.getTermInformation().getCurrentTerm(), newBehavior instanceof Follower); }
@Test public void testResponseToRequestVoteWithLowerTerm() { candidate = new Candidate(createActorContext()); setupPeers(1); candidate.handleMessage(peerActors[0], new RequestVote(1, "test", 0, 0)); RequestVoteReply reply = MessageCollectorActor.expectFirstMatching( peerActors[0], RequestVoteReply.class); assertEquals("isVoteGranted", false, reply.isVoteGranted()); assertEquals("getTerm", 2, reply.getTerm()); }
@Test public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForMatches() { MockRaftActorContext context = createActorContext(); context.getTermInformation().update(1000, null); candidate = new Candidate(context); setupPeers(1); candidate.handleMessage(peerActors[0], new RequestVote(1001, context.getId(), 10000, 999)); RequestVoteReply reply = MessageCollectorActor.expectFirstMatching( peerActors[0], RequestVoteReply.class); assertEquals("isVoteGranted", true, reply.isVoteGranted()); assertEquals("getTerm", 1001, reply.getTerm()); }
@Test public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForDoesNotMatch() { MockRaftActorContext context = createActorContext(); context.getTermInformation().update(1000, null); candidate = new Candidate(context); setupPeers(1); candidate.handleMessage(peerActors[0], new RequestVote(1001, "candidate2", 10000, 999)); RequestVoteReply reply = MessageCollectorActor.expectFirstMatching( peerActors[0], RequestVoteReply.class); assertEquals("isVoteGranted", false, reply.isVoteGranted()); assertEquals("getTerm", 1001, reply.getTerm()); }
@Test @Override public void testHandleAppendEntriesAddSameEntryToLog() { MockRaftActorContext context = createActorContext(); context.getTermInformation().update(2, "test"); MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("zero"); setLastLogEntry(context, 2, 0, payload); List<ReplicatedLogEntry> entries = new ArrayList<>(); entries.add(new SimpleReplicatedLogEntry(0, 2, payload)); final AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 2, -1, (short)0); behavior = createBehavior(context); context.getTermInformation().update(2, "test"); behavior.handleMessage(candidateActor, "unknown"); RaftActorBehavior raftBehavior = behavior.handleMessage(candidateActor, appendEntries); assertEquals("Raft state", RaftState.Follower, raftBehavior.state()); assertEquals("ReplicatedLog size", 1, context.getReplicatedLog().size()); handleAppendEntriesAddSameEntryToLogReply(candidateActor); }