@Test public void shouldIncreaseCurrentTermAndGrantVoteAndResetElectionTimeoutIfReceiveARequestVoteWithHigherTermWhileCurrentlyALeader() throws StorageException { becomeLeaderInTerm(3, false); long preRequestVoteHeartbeatTimeoutTick = getHeartbeatTimeoutTick(); long ticksBeforeRequestVoteIsReceived = (preRequestVoteHeartbeatTimeoutTick - timer.getTick()) / 2; timer.fastForward(ticksBeforeRequestVoteIsReceived); sender.drainSentRPCs(); algorithm.onRequestVote(S_01, 6, 1, 5); assertThatStateAfterRequestVoteIs(6, S_01, FOLLOWER); RequestVoteReply requestVoteReply = sender.nextAndRemove(RequestVoteReply.class); assertThatRequestVoteReplyHasValues(requestVoteReply, S_01, 6, true); assertThatNoMoreRPCsWereSent(); long postRequestVoteElectionTimeoutTick = getElectionTimeoutTick(); assertThat(postRequestVoteElectionTimeoutTick, greaterThan(preRequestVoteHeartbeatTimeoutTick)); assertThat(postRequestVoteElectionTimeoutTick, equalTo(timer.getTick() + RaftConstants.MIN_ELECTION_TIMEOUT)); assertThatLogContains( SENTINEL(), NOOP(1, 3) ); assertThatTermAndCommitIndexHaveValues(6, 0); }
@Override public synchronized void onRequestVote(String server, long term, long lastLogIndex, long lastLogTerm) { LOGGER.trace("{}: RequestVote from {}: term:{} lastLogIndex:{} lastLogTerm:{}", self, server, term, lastLogIndex, lastLogTerm); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(lastLogIndex >= 0); checkArgument(lastLogTerm >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.trace("{}: RequestVote from {}: old term: term:{} currentTerm:{}", self, server, term, currentTerm); sendRequestVoteReplyIgnoreRPCException(server, term, currentTerm, false); return; } if (term > currentTerm) { String votedFor = store.getVotedFor(term); checkState(votedFor == null, "voted in future term:%s for:%s", term, votedFor); becomeFollower(term, null); currentTerm = store.getCurrentTerm(); } checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); String votedFor = store.getVotedFor(term); LogEntry selfLastLog = checkNotNull(log.getLast()); int candidateLogDominates = doesLogDominate(lastLogIndex, lastLogTerm, selfLastLog.getIndex(), selfLastLog.getTerm()); boolean voteGranted = false; if (leader == null && ((votedFor == null && candidateLogDominates >= 0) || (self.equals(votedFor) && candidateLogDominates == 1))) { becomeFollowerWithoutUpdatingCurrentTerm(currentTerm, null); store.setVotedFor(term, server); voteGranted = true; } sendRequestVoteReplyIgnoreRPCException(server, currentTerm, currentTerm, voteGranted); } catch (StorageException e) { handleStorageException(e); } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onRequestVote(String server, long term, long lastLogIndex, long lastLogTerm) { LOGGER.trace("{}: RequestVote from {}: term:{} lastLogIndex:{} lastLogTerm:{}", self, server, term, lastLogIndex, lastLogTerm); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(lastLogIndex >= 0); checkArgument(lastLogTerm >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.trace("{}: RequestVote from {}: old term: term:{} currentTerm:{}", self, server, term, currentTerm); sendRequestVoteReplyIgnoreRPCException(server, term, currentTerm, false); return; } if (term > currentTerm) { String votedFor = store.getVotedFor(term); checkState(votedFor == null, "voted in future term:%s for:%s", term, votedFor); becomeFollower(term, null); currentTerm = store.getCurrentTerm(); } checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); String votedFor = store.getVotedFor(term); LogEntry selfLastLog = checkNotNull(log.getLast()); int candidateLogDominates = doesLogDominate(lastLogIndex, lastLogTerm, selfLastLog.getIndex(), selfLastLog.getTerm()); boolean voteGranted = false; if (leader == null && ((votedFor == null && candidateLogDominates >= 0) || (self.equals(votedFor) && candidateLogDominates == 1))) { becomeFollowerWithoutUpdatingCurrentTerm(currentTerm, null); store.setVotedFor(term, server); voteGranted = true; } sendRequestVoteReplyIgnoreRPCException(server, currentTerm, currentTerm, voteGranted); } catch (StorageException e) { handleStorageException(e); } } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onRequestVote(String server, long term, long lastLogIndex, long lastLogTerm) { LOGGER.trace("{}: RequestVote from {}: term:{} lastLogIndex:{} lastLogTerm:{}", self, server, term, lastLogIndex, lastLogTerm); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(lastLogIndex >= 0); checkArgument(lastLogTerm >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.trace("{}: RequestVote from {}: old term: term:{} currentTerm:{}", self, server, term, currentTerm); sendRequestVoteReplyIgnoreRPCException(server, term, currentTerm, false); return; } if (term > currentTerm) { String votedFor = store.getVotedFor(term); checkState(votedFor == null, "voted in future term:%s for:%s", term, votedFor); becomeFollower(term, null); currentTerm = store.getCurrentTerm(); } checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); String votedFor = store.getVotedFor(term); LogEntry selfLastLog = checkNotNull(log.getLast()); int candidateLogDominates = doesLogDominate(lastLogIndex, lastLogTerm, selfLastLog.getIndex(), selfLastLog.getTerm()); boolean voteGranted = false; if (leader == null && ((votedFor == null && candidateLogDominates >= 0) || (self.equals(votedFor) && candidateLogDominates == 1))) { becomeFollowerWithoutUpdatingCurrentTerm(currentTerm, null); store.setVotedFor(term, server); voteGranted = true; } sendRequestVoteReplyIgnoreRPCException(server, currentTerm, currentTerm, voteGranted); } catch (StorageException e) { handleStorageException(e); } } RaftAlgorithm(
            Random random,
            Timer timer,
            RPCSender sender,
            Store store,
            Log log,
            SnapshotsStore snapshotsStore,
            RaftListener listener,
            String self,
            Set<String> cluster);  RaftAlgorithm(
            Random random,
            Timer timer,
            RPCSender sender,
            Store store,
            Log log,
            SnapshotsStore snapshotsStore,
            RaftListener listener,
            String self,
            Set<String> cluster,
            int minEntriesToSnapshot,
            long snapshotCheckInterval,
            long rpcTimeout,
            long minElectionTimeout,
            long additionalElectionTimeoutRange,
            long heartbeatInterval,
            TimeUnit timeoutTimeUnit); }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onRequestVote(String server, long term, long lastLogIndex, long lastLogTerm) { LOGGER.trace("{}: RequestVote from {}: term:{} lastLogIndex:{} lastLogTerm:{}", self, server, term, lastLogIndex, lastLogTerm); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(lastLogIndex >= 0); checkArgument(lastLogTerm >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.trace("{}: RequestVote from {}: old term: term:{} currentTerm:{}", self, server, term, currentTerm); sendRequestVoteReplyIgnoreRPCException(server, term, currentTerm, false); return; } if (term > currentTerm) { String votedFor = store.getVotedFor(term); checkState(votedFor == null, "voted in future term:%s for:%s", term, votedFor); becomeFollower(term, null); currentTerm = store.getCurrentTerm(); } checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); String votedFor = store.getVotedFor(term); LogEntry selfLastLog = checkNotNull(log.getLast()); int candidateLogDominates = doesLogDominate(lastLogIndex, lastLogTerm, selfLastLog.getIndex(), selfLastLog.getTerm()); boolean voteGranted = false; if (leader == null && ((votedFor == null && candidateLogDominates >= 0) || (self.equals(votedFor) && candidateLogDominates == 1))) { becomeFollowerWithoutUpdatingCurrentTerm(currentTerm, null); store.setVotedFor(term, server); voteGranted = true; } sendRequestVoteReplyIgnoreRPCException(server, currentTerm, currentTerm, voteGranted); } catch (StorageException e) { handleStorageException(e); } } RaftAlgorithm(
            Random random,
            Timer timer,
            RPCSender sender,
            Store store,
            Log log,
            SnapshotsStore snapshotsStore,
            RaftListener listener,
            String self,
            Set<String> cluster);  RaftAlgorithm(
            Random random,
            Timer timer,
            RPCSender sender,
            Store store,
            Log log,
            SnapshotsStore snapshotsStore,
            RaftListener listener,
            String self,
            Set<String> cluster,
            int minEntriesToSnapshot,
            long snapshotCheckInterval,
            long rpcTimeout,
            long minElectionTimeout,
            long additionalElectionTimeoutRange,
            long heartbeatInterval,
            TimeUnit timeoutTimeUnit); synchronized void initialize(); synchronized void start(); synchronized void stop(); @Override synchronized void onRequestVote(String server, long term, long lastLogIndex, long lastLogTerm); @Override synchronized void onRequestVoteReply(String server, long term, boolean voteGranted); @Override synchronized void onAppendEntries(String server, long term, long commitIndex, long prevLogIndex, long prevLogTerm, @Nullable Collection<LogEntry> entries); @Override synchronized void onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied); @Override synchronized void snapshotWritten(SnapshotWriter snapshotWriter); @Override synchronized @Nullable Committed getNextCommitted(long indexToSearchFrom); @Override synchronized ListenableFuture<Void> submitCommand(Command command); }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onRequestVote(String server, long term, long lastLogIndex, long lastLogTerm) { LOGGER.trace("{}: RequestVote from {}: term:{} lastLogIndex:{} lastLogTerm:{}", self, server, term, lastLogIndex, lastLogTerm); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(lastLogIndex >= 0); checkArgument(lastLogTerm >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.trace("{}: RequestVote from {}: old term: term:{} currentTerm:{}", self, server, term, currentTerm); sendRequestVoteReplyIgnoreRPCException(server, term, currentTerm, false); return; } if (term > currentTerm) { String votedFor = store.getVotedFor(term); checkState(votedFor == null, "voted in future term:%s for:%s", term, votedFor); becomeFollower(term, null); currentTerm = store.getCurrentTerm(); } checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); String votedFor = store.getVotedFor(term); LogEntry selfLastLog = checkNotNull(log.getLast()); int candidateLogDominates = doesLogDominate(lastLogIndex, lastLogTerm, selfLastLog.getIndex(), selfLastLog.getTerm()); boolean voteGranted = false; if (leader == null && ((votedFor == null && candidateLogDominates >= 0) || (self.equals(votedFor) && candidateLogDominates == 1))) { becomeFollowerWithoutUpdatingCurrentTerm(currentTerm, null); store.setVotedFor(term, server); voteGranted = true; } sendRequestVoteReplyIgnoreRPCException(server, currentTerm, currentTerm, voteGranted); } catch (StorageException e) { handleStorageException(e); } } RaftAlgorithm(
            Random random,
            Timer timer,
            RPCSender sender,
            Store store,
            Log log,
            SnapshotsStore snapshotsStore,
            RaftListener listener,
            String self,
            Set<String> cluster);  RaftAlgorithm(
            Random random,
            Timer timer,
            RPCSender sender,
            Store store,
            Log log,
            SnapshotsStore snapshotsStore,
            RaftListener listener,
            String self,
            Set<String> cluster,
            int minEntriesToSnapshot,
            long snapshotCheckInterval,
            long rpcTimeout,
            long minElectionTimeout,
            long additionalElectionTimeoutRange,
            long heartbeatInterval,
            TimeUnit timeoutTimeUnit); synchronized void initialize(); synchronized void start(); synchronized void stop(); @Override synchronized void onRequestVote(String server, long term, long lastLogIndex, long lastLogTerm); @Override synchronized void onRequestVoteReply(String server, long term, boolean voteGranted); @Override synchronized void onAppendEntries(String server, long term, long commitIndex, long prevLogIndex, long prevLogTerm, @Nullable Collection<LogEntry> entries); @Override synchronized void onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied); @Override synchronized void snapshotWritten(SnapshotWriter snapshotWriter); @Override synchronized @Nullable Committed getNextCommitted(long indexToSearchFrom); @Override synchronized ListenableFuture<Void> submitCommand(Command command);  }