@Test public void shouldResetElectionTimeoutAndSetSendingServerAsLeaderOnReceivingAnAppendEntriesForElectionTerm() throws StorageException { store.setCurrentTerm(3); long preAppendEntriesElectionTimeoutTick = getElectionTimeoutTick(); assertThatSelfTransitionedToFollower(3, 0, null, false); long ticksBeforeAppendEntriesReceived = (preAppendEntriesElectionTimeoutTick - timer.getTick()) / 2; timer.fastForward(ticksBeforeAppendEntriesReceived); algorithm.onAppendEntries(S_01, 3, 0, 0, 0, null); assertThatSelfTransitionedToFollower(3, 0, S_01, true); long postAppendEntriesElectionTimeoutTick = getElectionTimeoutTick(); assertThat(postAppendEntriesElectionTimeoutTick, greaterThan(preAppendEntriesElectionTimeoutTick)); assertThat(postAppendEntriesElectionTimeoutTick, equalTo(timer.getTick() + RaftConstants.MIN_ELECTION_TIMEOUT)); assertThatLogContainsOnlySentinel(); assertThatTermAndCommitIndexHaveValues(3, 0); }
@Override public synchronized void onAppendEntries(String server, long term, long commitIndex, long prevLogIndex, long prevLogTerm, @Nullable Collection<LogEntry> entries) { LOGGER.trace("{}: AppendEntries from {}: term:{} commitIndex:{} prevLogIndex:{} prevLogTerm:{} entryCount:{}", self, server, term, commitIndex, prevLogIndex, prevLogTerm, getEntryCount(entries)); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(commitIndex >= 0); checkArgument(prevLogIndex >= 0); checkArgument(prevLogTerm >= 0); checkArgument(entries == null || entries.size() > 0); if (entries == null) { entries = Collections.emptyList(); } long entryCount = entries.size(); long currentTerm = store.getCurrentTerm(); long selfCommitIndex = store.getCommitIndex(); if (term < currentTerm) { if (leader != null && leader.equals(server)) { LOGGER.trace("{}: AppendEntries from {}: late request", self, server); return; } LOGGER.trace("{}: AppendEntries from {}: term < currentTerm: term:{} currentTerm:{}", self, server, term, currentTerm); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } if (term > currentTerm) { becomeFollower(term, server); currentTerm = store.getCurrentTerm(); } if (role == Role.FOLLOWER && leader == null) { checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } if (role != Role.FOLLOWER) { checkState(role == Role.CANDIDATE); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } scheduleNextElectionTimeout(); LogEntry prevLog = log.get(prevLogIndex); if (prefixMismatch(prevLog, prevLogTerm)) { LOGGER.trace("{}: prefix mismatch at index:{} expected term:{} entry:{}", self, prevLogIndex, prevLogTerm, prevLog); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } else { prevLog = checkNotNull(prevLog); if (nextToApplyLogIndex == -1) { nextToApplyLogIndex = Math.max(selfCommitIndex + 1, prevLogIndex + 1); } } log.truncate(nextToApplyLogIndex); for (LogEntry entry : entries) { entry = checkNotNull(entry); checkArgument(entry.getIndex() == prevLog.getIndex() + 1, "entries has hole: entry:%s prevLog:%s", entry, prevLog); prevLog = entry; if (entry.getIndex() != nextToApplyLogIndex) { LogEntry skippedLog = log.get(entry.getIndex()); checkState(entry.equals(skippedLog), "mismatch at index:%s entry:%s logEntry:%s", entry.getIndex(), entry, skippedLog); } else { LOGGER.trace("{}: add entry:{}", self, entry); log.put(entry); nextToApplyLogIndex = nextToApplyLogIndex + 1; } } sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, true); long originalSelfCommitIndex = selfCommitIndex; long possibleSelfCommitIndex = Math.min(nextToApplyLogIndex -1 , commitIndex); if (possibleSelfCommitIndex > selfCommitIndex) { store.setCommitIndex(possibleSelfCommitIndex); selfCommitIndex = store.getCommitIndex(); setCommandFuturesAndNotifyClient(originalSelfCommitIndex + 1, selfCommitIndex); } } catch (StorageException e) { handleStorageException(e); } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onAppendEntries(String server, long term, long commitIndex, long prevLogIndex, long prevLogTerm, @Nullable Collection<LogEntry> entries) { LOGGER.trace("{}: AppendEntries from {}: term:{} commitIndex:{} prevLogIndex:{} prevLogTerm:{} entryCount:{}", self, server, term, commitIndex, prevLogIndex, prevLogTerm, getEntryCount(entries)); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(commitIndex >= 0); checkArgument(prevLogIndex >= 0); checkArgument(prevLogTerm >= 0); checkArgument(entries == null || entries.size() > 0); if (entries == null) { entries = Collections.emptyList(); } long entryCount = entries.size(); long currentTerm = store.getCurrentTerm(); long selfCommitIndex = store.getCommitIndex(); if (term < currentTerm) { if (leader != null && leader.equals(server)) { LOGGER.trace("{}: AppendEntries from {}: late request", self, server); return; } LOGGER.trace("{}: AppendEntries from {}: term < currentTerm: term:{} currentTerm:{}", self, server, term, currentTerm); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } if (term > currentTerm) { becomeFollower(term, server); currentTerm = store.getCurrentTerm(); } if (role == Role.FOLLOWER && leader == null) { checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } if (role != Role.FOLLOWER) { checkState(role == Role.CANDIDATE); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } scheduleNextElectionTimeout(); LogEntry prevLog = log.get(prevLogIndex); if (prefixMismatch(prevLog, prevLogTerm)) { LOGGER.trace("{}: prefix mismatch at index:{} expected term:{} entry:{}", self, prevLogIndex, prevLogTerm, prevLog); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } else { prevLog = checkNotNull(prevLog); if (nextToApplyLogIndex == -1) { nextToApplyLogIndex = Math.max(selfCommitIndex + 1, prevLogIndex + 1); } } log.truncate(nextToApplyLogIndex); for (LogEntry entry : entries) { entry = checkNotNull(entry); checkArgument(entry.getIndex() == prevLog.getIndex() + 1, "entries has hole: entry:%s prevLog:%s", entry, prevLog); prevLog = entry; if (entry.getIndex() != nextToApplyLogIndex) { LogEntry skippedLog = log.get(entry.getIndex()); checkState(entry.equals(skippedLog), "mismatch at index:%s entry:%s logEntry:%s", entry.getIndex(), entry, skippedLog); } else { LOGGER.trace("{}: add entry:{}", self, entry); log.put(entry); nextToApplyLogIndex = nextToApplyLogIndex + 1; } } sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, true); long originalSelfCommitIndex = selfCommitIndex; long possibleSelfCommitIndex = Math.min(nextToApplyLogIndex -1 , commitIndex); if (possibleSelfCommitIndex > selfCommitIndex) { store.setCommitIndex(possibleSelfCommitIndex); selfCommitIndex = store.getCommitIndex(); setCommandFuturesAndNotifyClient(originalSelfCommitIndex + 1, selfCommitIndex); } } catch (StorageException e) { handleStorageException(e); } } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onAppendEntries(String server, long term, long commitIndex, long prevLogIndex, long prevLogTerm, @Nullable Collection<LogEntry> entries) { LOGGER.trace("{}: AppendEntries from {}: term:{} commitIndex:{} prevLogIndex:{} prevLogTerm:{} entryCount:{}", self, server, term, commitIndex, prevLogIndex, prevLogTerm, getEntryCount(entries)); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(commitIndex >= 0); checkArgument(prevLogIndex >= 0); checkArgument(prevLogTerm >= 0); checkArgument(entries == null || entries.size() > 0); if (entries == null) { entries = Collections.emptyList(); } long entryCount = entries.size(); long currentTerm = store.getCurrentTerm(); long selfCommitIndex = store.getCommitIndex(); if (term < currentTerm) { if (leader != null && leader.equals(server)) { LOGGER.trace("{}: AppendEntries from {}: late request", self, server); return; } LOGGER.trace("{}: AppendEntries from {}: term < currentTerm: term:{} currentTerm:{}", self, server, term, currentTerm); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } if (term > currentTerm) { becomeFollower(term, server); currentTerm = store.getCurrentTerm(); } if (role == Role.FOLLOWER && leader == null) { checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } if (role != Role.FOLLOWER) { checkState(role == Role.CANDIDATE); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } scheduleNextElectionTimeout(); LogEntry prevLog = log.get(prevLogIndex); if (prefixMismatch(prevLog, prevLogTerm)) { LOGGER.trace("{}: prefix mismatch at index:{} expected term:{} entry:{}", self, prevLogIndex, prevLogTerm, prevLog); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } else { prevLog = checkNotNull(prevLog); if (nextToApplyLogIndex == -1) { nextToApplyLogIndex = Math.max(selfCommitIndex + 1, prevLogIndex + 1); } } log.truncate(nextToApplyLogIndex); for (LogEntry entry : entries) { entry = checkNotNull(entry); checkArgument(entry.getIndex() == prevLog.getIndex() + 1, "entries has hole: entry:%s prevLog:%s", entry, prevLog); prevLog = entry; if (entry.getIndex() != nextToApplyLogIndex) { LogEntry skippedLog = log.get(entry.getIndex()); checkState(entry.equals(skippedLog), "mismatch at index:%s entry:%s logEntry:%s", entry.getIndex(), entry, skippedLog); } else { LOGGER.trace("{}: add entry:{}", self, entry); log.put(entry); nextToApplyLogIndex = nextToApplyLogIndex + 1; } } sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, true); long originalSelfCommitIndex = selfCommitIndex; long possibleSelfCommitIndex = Math.min(nextToApplyLogIndex -1 , commitIndex); if (possibleSelfCommitIndex > selfCommitIndex) { store.setCommitIndex(possibleSelfCommitIndex); selfCommitIndex = store.getCommitIndex(); setCommandFuturesAndNotifyClient(originalSelfCommitIndex + 1, selfCommitIndex); } } 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 onAppendEntries(String server, long term, long commitIndex, long prevLogIndex, long prevLogTerm, @Nullable Collection<LogEntry> entries) { LOGGER.trace("{}: AppendEntries from {}: term:{} commitIndex:{} prevLogIndex:{} prevLogTerm:{} entryCount:{}", self, server, term, commitIndex, prevLogIndex, prevLogTerm, getEntryCount(entries)); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(commitIndex >= 0); checkArgument(prevLogIndex >= 0); checkArgument(prevLogTerm >= 0); checkArgument(entries == null || entries.size() > 0); if (entries == null) { entries = Collections.emptyList(); } long entryCount = entries.size(); long currentTerm = store.getCurrentTerm(); long selfCommitIndex = store.getCommitIndex(); if (term < currentTerm) { if (leader != null && leader.equals(server)) { LOGGER.trace("{}: AppendEntries from {}: late request", self, server); return; } LOGGER.trace("{}: AppendEntries from {}: term < currentTerm: term:{} currentTerm:{}", self, server, term, currentTerm); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } if (term > currentTerm) { becomeFollower(term, server); currentTerm = store.getCurrentTerm(); } if (role == Role.FOLLOWER && leader == null) { checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } if (role != Role.FOLLOWER) { checkState(role == Role.CANDIDATE); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } scheduleNextElectionTimeout(); LogEntry prevLog = log.get(prevLogIndex); if (prefixMismatch(prevLog, prevLogTerm)) { LOGGER.trace("{}: prefix mismatch at index:{} expected term:{} entry:{}", self, prevLogIndex, prevLogTerm, prevLog); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } else { prevLog = checkNotNull(prevLog); if (nextToApplyLogIndex == -1) { nextToApplyLogIndex = Math.max(selfCommitIndex + 1, prevLogIndex + 1); } } log.truncate(nextToApplyLogIndex); for (LogEntry entry : entries) { entry = checkNotNull(entry); checkArgument(entry.getIndex() == prevLog.getIndex() + 1, "entries has hole: entry:%s prevLog:%s", entry, prevLog); prevLog = entry; if (entry.getIndex() != nextToApplyLogIndex) { LogEntry skippedLog = log.get(entry.getIndex()); checkState(entry.equals(skippedLog), "mismatch at index:%s entry:%s logEntry:%s", entry.getIndex(), entry, skippedLog); } else { LOGGER.trace("{}: add entry:{}", self, entry); log.put(entry); nextToApplyLogIndex = nextToApplyLogIndex + 1; } } sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, true); long originalSelfCommitIndex = selfCommitIndex; long possibleSelfCommitIndex = Math.min(nextToApplyLogIndex -1 , commitIndex); if (possibleSelfCommitIndex > selfCommitIndex) { store.setCommitIndex(possibleSelfCommitIndex); selfCommitIndex = store.getCommitIndex(); setCommandFuturesAndNotifyClient(originalSelfCommitIndex + 1, selfCommitIndex); } } 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 onAppendEntries(String server, long term, long commitIndex, long prevLogIndex, long prevLogTerm, @Nullable Collection<LogEntry> entries) { LOGGER.trace("{}: AppendEntries from {}: term:{} commitIndex:{} prevLogIndex:{} prevLogTerm:{} entryCount:{}", self, server, term, commitIndex, prevLogIndex, prevLogTerm, getEntryCount(entries)); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(commitIndex >= 0); checkArgument(prevLogIndex >= 0); checkArgument(prevLogTerm >= 0); checkArgument(entries == null || entries.size() > 0); if (entries == null) { entries = Collections.emptyList(); } long entryCount = entries.size(); long currentTerm = store.getCurrentTerm(); long selfCommitIndex = store.getCommitIndex(); if (term < currentTerm) { if (leader != null && leader.equals(server)) { LOGGER.trace("{}: AppendEntries from {}: late request", self, server); return; } LOGGER.trace("{}: AppendEntries from {}: term < currentTerm: term:{} currentTerm:{}", self, server, term, currentTerm); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } if (term > currentTerm) { becomeFollower(term, server); currentTerm = store.getCurrentTerm(); } if (role == Role.FOLLOWER && leader == null) { checkState(term == currentTerm, "term:%s currentTerm:%s", term, currentTerm); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } if (role != Role.FOLLOWER) { checkState(role == Role.CANDIDATE); becomeFollowerWithoutUpdatingCurrentTerm(term, server); } scheduleNextElectionTimeout(); LogEntry prevLog = log.get(prevLogIndex); if (prefixMismatch(prevLog, prevLogTerm)) { LOGGER.trace("{}: prefix mismatch at index:{} expected term:{} entry:{}", self, prevLogIndex, prevLogTerm, prevLog); sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, false); return; } else { prevLog = checkNotNull(prevLog); if (nextToApplyLogIndex == -1) { nextToApplyLogIndex = Math.max(selfCommitIndex + 1, prevLogIndex + 1); } } log.truncate(nextToApplyLogIndex); for (LogEntry entry : entries) { entry = checkNotNull(entry); checkArgument(entry.getIndex() == prevLog.getIndex() + 1, "entries has hole: entry:%s prevLog:%s", entry, prevLog); prevLog = entry; if (entry.getIndex() != nextToApplyLogIndex) { LogEntry skippedLog = log.get(entry.getIndex()); checkState(entry.equals(skippedLog), "mismatch at index:%s entry:%s logEntry:%s", entry.getIndex(), entry, skippedLog); } else { LOGGER.trace("{}: add entry:{}", self, entry); log.put(entry); nextToApplyLogIndex = nextToApplyLogIndex + 1; } } sendAppendEntriesReplyIgnoreRPCException(server, currentTerm, prevLogIndex, entryCount, true); long originalSelfCommitIndex = selfCommitIndex; long possibleSelfCommitIndex = Math.min(nextToApplyLogIndex -1 , commitIndex); if (possibleSelfCommitIndex > selfCommitIndex) { store.setCommitIndex(possibleSelfCommitIndex); selfCommitIndex = store.getCommitIndex(); setCommandFuturesAndNotifyClient(originalSelfCommitIndex + 1, selfCommitIndex); } } 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);  }