@Test public void shouldInitializeAsLeaderCorrectlyAndCommitLeaderNoopIfQuorumOfAcksReceived() throws StorageException, RPCException { assertThatLogContains( SENTINEL() ); becomeLeaderInTerm(3, false); Collection<AppendEntries> heartbeats = getRPCs(4, AppendEntries.class); assertThatAppendEntriesHaveValues(heartbeats, 3, 0, 0, 0, NOOP(1, 3)); assertThatNoMoreRPCsWereSent(); assertThatLogContains( SENTINEL(), NOOP(1, 3) ); assertThatTermAndCommitIndexHaveValues(3, 0); algorithm.onAppendEntriesReply(S_01, 3, 0, 1, true); algorithm.onAppendEntriesReply(S_02, 3, 0, 1, true); assertThatLogContains( SENTINEL(), NOOP(1, 3) ); assertThatTermAndCommitIndexHaveValues(3, 1); long heartbeatTimeout = getHeartbeatTimeoutTick(); assertThat(heartbeatTimeout, equalTo((long) RaftConstants.HEARTBEAT_INTERVAL)); }
@Override public synchronized void onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied) { LOGGER.trace("{}: AppendEntriesReply from {}: term:{} prevLogIndex:{} entryCount:{} applied:{}", self, server, term, prevLogIndex, entryCount, applied); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(prevLogIndex >= 0); checkArgument(entryCount >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: ignore: term:{} currentTerm:{}", self, server, term, currentTerm); return; } if (term > currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: become follower: term:{} currentTerm:{}", self, server, term, currentTerm); becomeFollower(term, null); return; } checkState(role == Role.LEADER, "role:%s", role); checkState(self.equals(leader), "self:%s leader:%s", self, leader); ServerDatum serverDatum = checkNotNull(serverData.get(server), "no server data for %s", server); if (serverDatum.phase == Phase.PREFIX_SEARCH) { if (prevLogIndex + 1 == serverDatum.nextIndex) { if (applied) { serverDatum.phase = Phase.APPLYING; } else { checkArgument(prevLogIndex > 0); serverDatum.nextIndex--; return; } } else { return; } } checkState(serverDatum.phase == Phase.APPLYING); checkState(applied); long lastAppliedIndex = prevLogIndex + entryCount; LogEntry lastLog = checkNotNull(log.getLast()); checkArgument(lastAppliedIndex <= lastLog.getIndex(), "lastAppliedIndex:%s lastLog:%s", lastAppliedIndex, lastLog); if (lastAppliedIndex >= serverDatum.nextIndex) { serverDatum.nextIndex = lastAppliedIndex + 1; } else { return; } long originalCommitIndex = store.getCommitIndex(); long possibleCommitIndex = findPossibleCommitIndex(lastLog, originalCommitIndex); if (possibleCommitIndex > originalCommitIndex) { LogEntry potentialCommittedLog = checkNotNull(log.get(possibleCommitIndex), "possibleCommitIndex:%s", possibleCommitIndex); checkState(potentialCommittedLog.getTerm() <= currentTerm, "future log:%s", potentialCommittedLog); if (potentialCommittedLog.getTerm() == currentTerm) { store.setCommitIndex(possibleCommitIndex); setCommandFuturesAndNotifyClient(originalCommitIndex + 1, possibleCommitIndex); } } } catch (StorageException e) { handleStorageException(e); } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied) { LOGGER.trace("{}: AppendEntriesReply from {}: term:{} prevLogIndex:{} entryCount:{} applied:{}", self, server, term, prevLogIndex, entryCount, applied); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(prevLogIndex >= 0); checkArgument(entryCount >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: ignore: term:{} currentTerm:{}", self, server, term, currentTerm); return; } if (term > currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: become follower: term:{} currentTerm:{}", self, server, term, currentTerm); becomeFollower(term, null); return; } checkState(role == Role.LEADER, "role:%s", role); checkState(self.equals(leader), "self:%s leader:%s", self, leader); ServerDatum serverDatum = checkNotNull(serverData.get(server), "no server data for %s", server); if (serverDatum.phase == Phase.PREFIX_SEARCH) { if (prevLogIndex + 1 == serverDatum.nextIndex) { if (applied) { serverDatum.phase = Phase.APPLYING; } else { checkArgument(prevLogIndex > 0); serverDatum.nextIndex--; return; } } else { return; } } checkState(serverDatum.phase == Phase.APPLYING); checkState(applied); long lastAppliedIndex = prevLogIndex + entryCount; LogEntry lastLog = checkNotNull(log.getLast()); checkArgument(lastAppliedIndex <= lastLog.getIndex(), "lastAppliedIndex:%s lastLog:%s", lastAppliedIndex, lastLog); if (lastAppliedIndex >= serverDatum.nextIndex) { serverDatum.nextIndex = lastAppliedIndex + 1; } else { return; } long originalCommitIndex = store.getCommitIndex(); long possibleCommitIndex = findPossibleCommitIndex(lastLog, originalCommitIndex); if (possibleCommitIndex > originalCommitIndex) { LogEntry potentialCommittedLog = checkNotNull(log.get(possibleCommitIndex), "possibleCommitIndex:%s", possibleCommitIndex); checkState(potentialCommittedLog.getTerm() <= currentTerm, "future log:%s", potentialCommittedLog); if (potentialCommittedLog.getTerm() == currentTerm) { store.setCommitIndex(possibleCommitIndex); setCommandFuturesAndNotifyClient(originalCommitIndex + 1, possibleCommitIndex); } } } catch (StorageException e) { handleStorageException(e); } } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized void onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied) { LOGGER.trace("{}: AppendEntriesReply from {}: term:{} prevLogIndex:{} entryCount:{} applied:{}", self, server, term, prevLogIndex, entryCount, applied); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(prevLogIndex >= 0); checkArgument(entryCount >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: ignore: term:{} currentTerm:{}", self, server, term, currentTerm); return; } if (term > currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: become follower: term:{} currentTerm:{}", self, server, term, currentTerm); becomeFollower(term, null); return; } checkState(role == Role.LEADER, "role:%s", role); checkState(self.equals(leader), "self:%s leader:%s", self, leader); ServerDatum serverDatum = checkNotNull(serverData.get(server), "no server data for %s", server); if (serverDatum.phase == Phase.PREFIX_SEARCH) { if (prevLogIndex + 1 == serverDatum.nextIndex) { if (applied) { serverDatum.phase = Phase.APPLYING; } else { checkArgument(prevLogIndex > 0); serverDatum.nextIndex--; return; } } else { return; } } checkState(serverDatum.phase == Phase.APPLYING); checkState(applied); long lastAppliedIndex = prevLogIndex + entryCount; LogEntry lastLog = checkNotNull(log.getLast()); checkArgument(lastAppliedIndex <= lastLog.getIndex(), "lastAppliedIndex:%s lastLog:%s", lastAppliedIndex, lastLog); if (lastAppliedIndex >= serverDatum.nextIndex) { serverDatum.nextIndex = lastAppliedIndex + 1; } else { return; } long originalCommitIndex = store.getCommitIndex(); long possibleCommitIndex = findPossibleCommitIndex(lastLog, originalCommitIndex); if (possibleCommitIndex > originalCommitIndex) { LogEntry potentialCommittedLog = checkNotNull(log.get(possibleCommitIndex), "possibleCommitIndex:%s", possibleCommitIndex); checkState(potentialCommittedLog.getTerm() <= currentTerm, "future log:%s", potentialCommittedLog); if (potentialCommittedLog.getTerm() == currentTerm) { store.setCommitIndex(possibleCommitIndex); setCommandFuturesAndNotifyClient(originalCommitIndex + 1, possibleCommitIndex); } } } 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 onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied) { LOGGER.trace("{}: AppendEntriesReply from {}: term:{} prevLogIndex:{} entryCount:{} applied:{}", self, server, term, prevLogIndex, entryCount, applied); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(prevLogIndex >= 0); checkArgument(entryCount >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: ignore: term:{} currentTerm:{}", self, server, term, currentTerm); return; } if (term > currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: become follower: term:{} currentTerm:{}", self, server, term, currentTerm); becomeFollower(term, null); return; } checkState(role == Role.LEADER, "role:%s", role); checkState(self.equals(leader), "self:%s leader:%s", self, leader); ServerDatum serverDatum = checkNotNull(serverData.get(server), "no server data for %s", server); if (serverDatum.phase == Phase.PREFIX_SEARCH) { if (prevLogIndex + 1 == serverDatum.nextIndex) { if (applied) { serverDatum.phase = Phase.APPLYING; } else { checkArgument(prevLogIndex > 0); serverDatum.nextIndex--; return; } } else { return; } } checkState(serverDatum.phase == Phase.APPLYING); checkState(applied); long lastAppliedIndex = prevLogIndex + entryCount; LogEntry lastLog = checkNotNull(log.getLast()); checkArgument(lastAppliedIndex <= lastLog.getIndex(), "lastAppliedIndex:%s lastLog:%s", lastAppliedIndex, lastLog); if (lastAppliedIndex >= serverDatum.nextIndex) { serverDatum.nextIndex = lastAppliedIndex + 1; } else { return; } long originalCommitIndex = store.getCommitIndex(); long possibleCommitIndex = findPossibleCommitIndex(lastLog, originalCommitIndex); if (possibleCommitIndex > originalCommitIndex) { LogEntry potentialCommittedLog = checkNotNull(log.get(possibleCommitIndex), "possibleCommitIndex:%s", possibleCommitIndex); checkState(potentialCommittedLog.getTerm() <= currentTerm, "future log:%s", potentialCommittedLog); if (potentialCommittedLog.getTerm() == currentTerm) { store.setCommitIndex(possibleCommitIndex); setCommandFuturesAndNotifyClient(originalCommitIndex + 1, possibleCommitIndex); } } } 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 onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied) { LOGGER.trace("{}: AppendEntriesReply from {}: term:{} prevLogIndex:{} entryCount:{} applied:{}", self, server, term, prevLogIndex, entryCount, applied); if (!running) { logNotRunning(); return; } try { checkArgument(term >= 1); checkArgument(prevLogIndex >= 0); checkArgument(entryCount >= 0); long currentTerm = store.getCurrentTerm(); if (term < currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: ignore: term:{} currentTerm:{}", self, server, term, currentTerm); return; } if (term > currentTerm) { LOGGER.warn("{}: AppendEntriesReply from {}: become follower: term:{} currentTerm:{}", self, server, term, currentTerm); becomeFollower(term, null); return; } checkState(role == Role.LEADER, "role:%s", role); checkState(self.equals(leader), "self:%s leader:%s", self, leader); ServerDatum serverDatum = checkNotNull(serverData.get(server), "no server data for %s", server); if (serverDatum.phase == Phase.PREFIX_SEARCH) { if (prevLogIndex + 1 == serverDatum.nextIndex) { if (applied) { serverDatum.phase = Phase.APPLYING; } else { checkArgument(prevLogIndex > 0); serverDatum.nextIndex--; return; } } else { return; } } checkState(serverDatum.phase == Phase.APPLYING); checkState(applied); long lastAppliedIndex = prevLogIndex + entryCount; LogEntry lastLog = checkNotNull(log.getLast()); checkArgument(lastAppliedIndex <= lastLog.getIndex(), "lastAppliedIndex:%s lastLog:%s", lastAppliedIndex, lastLog); if (lastAppliedIndex >= serverDatum.nextIndex) { serverDatum.nextIndex = lastAppliedIndex + 1; } else { return; } long originalCommitIndex = store.getCommitIndex(); long possibleCommitIndex = findPossibleCommitIndex(lastLog, originalCommitIndex); if (possibleCommitIndex > originalCommitIndex) { LogEntry potentialCommittedLog = checkNotNull(log.get(possibleCommitIndex), "possibleCommitIndex:%s", possibleCommitIndex); checkState(potentialCommittedLog.getTerm() <= currentTerm, "future log:%s", potentialCommittedLog); if (potentialCommittedLog.getTerm() == currentTerm) { store.setCommitIndex(possibleCommitIndex); setCommandFuturesAndNotifyClient(originalCommitIndex + 1, possibleCommitIndex); } } } 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);  }