Candidate extends BaseState { @Override public void init(@Nonnull final RaftStateContext ctx) { final RaftLog log = getLog(); log.currentTerm(log.currentTerm() + 1); log.votedFor(Optional.of(log.self())); LOGGER.debug("Election starting for term {}", log.currentTerm()); List<CompletableFuture<RequestVoteResponse>> responses = log.members().stream().map(member -> sendVoteRequest(ctx, member) ).collect(toList()); responses.add(SUCCESS_VOTE); electionResult = majorityResponse(responses, voteGranted()); long timeout = electionTimeout + (RAND.nextLong() % electionTimeout); electionTimer = DeadlineTimer.start(scheduler, () -> { LOGGER.debug("Election timeout"); ctx.setState(Candidate.this, CANDIDATE); }, timeout); electionResult.thenAccept(elected -> { checkNotNull(elected); if (elected) { ctx.setState(Candidate.this, LEADER); } }).exceptionally(t -> { if (!electionResult.isCancelled()) { LOGGER.debug("Election failed with exception:", t); } return null; }); } @Inject  Candidate(RaftLog log, @RaftExecutor Fiber scheduler,
            @ElectionTimeout long electionTimeout, Client client); @Override void init(@Nonnull final RaftStateContext ctx); @Override void destroy(RaftStateContext ctx);  }
@Test public void testRequestVoteWithNewerTerm() throws Exception { Candidate candidate = new Candidate(mockRaftLog, scheduler, 150, mockRaftClient); candidate.init(mockRaftStateContext); Replica mockCandidate = config.getReplica("other"); RequestVote request = RequestVote.newBuilder() .setCandidateId(mockCandidate.toString()) .setLastLogIndex(1L) .setLastLogTerm(1L) .setTerm(4L) .build(); candidate.requestVote(mockRaftStateContext, request); verify(mockRaftLog).currentTerm(3L); verify(mockRaftLog).currentTerm(4L); verify(mockRaftLog, times(2)).currentTerm(anyLong()); verify(mockRaftLog).votedFor(java.util.Optional.of(self)); verify(mockRaftLog).votedFor(java.util.Optional.of(mockCandidate)); verify(mockRaftLog, times(2)).votedFor(any(Optional.class)); verify(mockRaftLog, never()).commitIndex(anyLong()); verify(mockRaftStateContext, times(1)).setState(any(Candidate.class), eq(Raft.StateType.FOLLOWER)); verify(mockRaftStateContext, times(1)).setState(any(Candidate.class), eq(Raft.StateType.LEADER)); verifyZeroInteractions(mockRaftClient); }
@Test public void testRequestVoteWithOlderTerm() throws Exception { Candidate candidate = new Candidate(mockRaftLog, scheduler, 150, mockRaftClient); candidate.init(mockRaftStateContext); Replica mockCandidate = config.getReplica("other"); RequestVote request = RequestVote.newBuilder() .setCandidateId(mockCandidate.toString()) .setLastLogIndex(1L) .setLastLogTerm(1L) .setTerm(1L) .build(); candidate.requestVote(mockRaftStateContext, request); verify(mockRaftLog).currentTerm(3L); verify(mockRaftLog, times(1)).currentTerm(anyLong()); verify(mockRaftLog).votedFor(java.util.Optional.of(self)); verify(mockRaftLog, never()).votedFor(java.util.Optional.of(mockCandidate)); verify(mockRaftLog, times(1)).votedFor(any(Optional.class)); verify(mockRaftLog, never()).commitIndex(anyLong()); verify(mockRaftStateContext).setState(any(Candidate.class), eq(Raft.StateType.LEADER)); verifyZeroInteractions(mockRaftClient); }
@Test public void testRequestVoteWithSameTerm() throws Exception { Candidate candidate = new Candidate(mockRaftLog, scheduler, 150, mockRaftClient); candidate.init(mockRaftStateContext); Replica mockCandidate = config.getReplica("other"); RequestVote request = RequestVote.newBuilder() .setCandidateId(mockCandidate.toString()) .setLastLogIndex(1L) .setLastLogTerm(1L) .setTerm(2L) .build(); candidate.requestVote(mockRaftStateContext, request); verify(mockRaftLog).currentTerm(3L); verify(mockRaftLog, times(1)).currentTerm(anyLong()); verify(mockRaftLog).votedFor(java.util.Optional.of(self)); verify(mockRaftLog, times(1)).votedFor(java.util.Optional.of(mockCandidate)); verify(mockRaftLog, times(2)).votedFor(any(Optional.class)); verify(mockRaftLog, never()).commitIndex(anyLong()); verify(mockRaftStateContext).setState(any(State.class), any(RaftStateContext.StateType.class)); verifyZeroInteractions(mockRaftStateContext); verifyZeroInteractions(mockRaftClient); }
@Test public void testAppendEntriesWithNewerTerm() throws Exception { Candidate candidate = new Candidate(mockRaftLog, scheduler, 1, mockRaftClient); candidate.init(mockRaftStateContext); Replica mockLeader = config.getReplica("other"); AppendEntries request = AppendEntries.newBuilder() .setTerm(4L) .setLeaderId(mockLeader.toString()) .setPrevLogIndex(1L) .setPrevLogTerm(1L) .setCommitIndex(1L) .build(); candidate.appendEntries(mockRaftStateContext, request); verify(mockRaftLog).currentTerm(3L); verify(mockRaftLog).currentTerm(4L); verify(mockRaftLog, times(2)).currentTerm(anyLong()); verify(mockRaftLog).votedFor(java.util.Optional.of(self)); verify(mockRaftLog, times(1)).votedFor(any(Optional.class)); verify(mockRaftLog, times(1)).commitIndex(anyLong()); verify(mockRaftStateContext).setState(any(Candidate.class), eq(Raft.StateType.FOLLOWER)); verify(mockRaftStateContext).setState(any(Candidate.class), eq(Raft.StateType.LEADER)); verifyZeroInteractions(mockRaftClient); }
@Test public void testAppendEntriesWithOlderTerm() throws Exception { Candidate candidate = new Candidate(mockRaftLog, scheduler, 1, mockRaftClient); candidate.init(mockRaftStateContext); Replica mockLeader = config.getReplica("other"); AppendEntries request = AppendEntries.newBuilder() .setTerm(1L) .setLeaderId(mockLeader.toString()) .setPrevLogIndex(1L) .setPrevLogTerm(1L) .setCommitIndex(1L) .build(); candidate.appendEntries(mockRaftStateContext, request); verify(mockRaftLog).currentTerm(3L); verify(mockRaftLog, times(1)).currentTerm(anyLong()); verify(mockRaftLog).votedFor(java.util.Optional.of(self)); verify(mockRaftLog, times(1)).votedFor(any(Optional.class)); verify(mockRaftLog, never()).commitIndex(anyLong()); verify(mockRaftStateContext).setState(any(Candidate.class), eq(Raft.StateType.LEADER)); verifyZeroInteractions(mockRaftClient); }
@Test public void testAppendEntriesWithSameTerm() throws Exception { Candidate candidate = new Candidate(mockRaftLog, scheduler, 1, mockRaftClient); candidate.init(mockRaftStateContext); AppendEntries request = AppendEntries.newBuilder() .setTerm(2L) .setLeaderId("leader:1000") .setPrevLogIndex(1L) .setPrevLogTerm(1L) .setCommitIndex(1L) .build(); candidate.appendEntries(mockRaftStateContext, request); verify(mockRaftLog).currentTerm(3L); verify(mockRaftLog, times(1)).currentTerm(anyLong()); verify(mockRaftLog).votedFor(java.util.Optional.of(self)); verify(mockRaftLog, times(1)).votedFor(any(Optional.class)); verify(mockRaftLog, times(1)).commitIndex(anyLong()); verify(mockRaftStateContext).setState(any(Candidate.class), eq(LEADER)); verifyZeroInteractions(mockRaftClient); }