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);  }
@Test public void shouldIgnoreDuplicateDelayedAppendEntriesReplyForLowerTerm() throws StorageException, RPCException { insertIntoLog(NOOP(1, 2)); insertIntoLog(NOOP(2, 2)); becomeLeaderInTerm(3, false); Collection<AppendEntries> appendEntriesRequests; appendEntriesRequests = getRPCs(4, AppendEntries.class); assertThatAppendEntriesHaveValues(appendEntriesRequests, 3, 0, 2, 2, NOOP(3, 3)); assertThatNoMoreRPCsWereSent(); algorithm.onAppendEntriesReply(S_04, 3, 2, 1, false); algorithm.onAppendEntriesReply(S_04, 2, 1, 1, false); algorithm.onAppendEntriesReply(S_04, 2, 1, 1, false); algorithm.onAppendEntriesReply(S_04, 2, 1, 1, false); timer.fastForward(); appendEntriesRequests = getRPCs(4, AppendEntries.class); for (AppendEntries appendEntries : appendEntriesRequests) { if (appendEntries.server.equals(S_04)) { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 1, 2, NOOP(2, 2), NOOP(3, 3)); } else { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 2, 2, NOOP(3, 3)); } } assertThatNoMoreRPCsWereSent(); assertThatLogContains( SENTINEL(), NOOP(1, 2), NOOP(2, 2), NOOP(3, 3) ); assertThatTermAndCommitIndexHaveValues(3, 0); }
@Test public void shouldMarkEntryAsCommittedIfReceiveAQuorumOfAcks() throws RPCException, StorageException { setupLeaderForCommitUnitTests(); long heartbeatTimeout = getHeartbeatTimeoutTick(); long heartbeatInterval = heartbeatTimeout - timer.getTick(); timer.fastForward(heartbeatInterval / 2); algorithm.onAppendEntriesReply(S_03, 3, 3, 1, true); algorithm.onAppendEntriesReply(S_01, 3, 3, 6, true); assertThatTermAndCommitIndexHaveValues(3, 4); timer.fastForward(heartbeatInterval / 4); algorithm.onAppendEntriesReply(S_02, 3, 3, 2, true); assertThatTermAndCommitIndexHaveValues(3, 5); fastForwardToHeartbeat(); Collection<AppendEntries> heartbeats = getRPCs(4, AppendEntries.class); for (AppendEntries heartbeat : heartbeats) { if (heartbeat.server.equals(S_01)) { assertThatAppendEntriesHasValues(heartbeat, 3, 5, 9, 3); } else if (heartbeat.server.equals(S_02)) { assertThatAppendEntriesHasValues( heartbeat, 3, 5, 5, 3, NOOP(6, 3), NOOP(7, 3), NOOP(8, 3), NOOP(9, 3) ); } else if (heartbeat.server.equals(S_03)) { assertThatAppendEntriesHasValues( heartbeat, 3, 5, 4, 3, NOOP(5, 3), NOOP(6, 3), NOOP(7, 3), NOOP(8, 3), NOOP(9, 3) ); } else { assertThatAppendEntriesHasValues( heartbeat, 3, 5, 3, 2, NOOP(4, 3), NOOP(5, 3), NOOP(6, 3), NOOP(7, 3), NOOP(8, 3), NOOP(9, 3) ); } } assertThatNoMoreRPCsWereSent(); assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3), NOOP(5, 3), NOOP(6, 3), NOOP(7, 3), NOOP(8, 3), NOOP(9, 3) ); assertThatTermAndCommitIndexHaveValues(3, 5); }
@Test public void shouldNotMarkEntryAsCommittedIfNotReceiveAQuorumOfAcks() throws RPCException, StorageException { setupLeaderForCommitUnitTests(); algorithm.onAppendEntriesReply(S_03, 3, 3, 1, true); timer.fastForward(); Collection<AppendEntries> heartbeats = getRPCs(4, AppendEntries.class); for (AppendEntries heartbeat : heartbeats) { if (heartbeat.server.equals(S_03)) { assertThatAppendEntriesHasValues( heartbeat, 3, 0, 4, 3, NOOP(5, 3), NOOP(6, 3), NOOP(7, 3), NOOP(8, 3), NOOP(9, 3) ); } else { assertThatAppendEntriesHasValues( heartbeat, 3, 0, 3, 2, NOOP(4, 3), NOOP(5, 3), NOOP(6, 3), NOOP(7, 3), NOOP(8, 3), NOOP(9, 3) ); } } assertThatNoMoreRPCsWereSent(); assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3), NOOP(5, 3), NOOP(6, 3), NOOP(7, 3), NOOP(8, 3), NOOP(9, 3) ); assertThatTermAndCommitIndexHaveValues(3, 0); }
@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)); }
@Test public void shouldDecrementNextIndexWhenReceivingAppendEntriesRejectToFindMatchingPrefix() throws StorageException, RPCException { insertIntoLog( NOOP(1, 1), NOOP(2, 1), NOOP(3, 2) ); store.setCommitIndex(1); becomeLeaderInTerm(3, false); assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3) ); Collection<AppendEntries> appendEntriesRequests; appendEntriesRequests = getRPCs(4, AppendEntries.class); assertThatAppendEntriesHaveValues(appendEntriesRequests, 3, 1, 3, 2, NOOP(4, 3)); assertThatNoMoreRPCsWereSent(); algorithm.onAppendEntriesReply(S_03, 3, 3, 1, false); timer.fastForward(); appendEntriesRequests = getRPCs(4, AppendEntries.class); for (AppendEntries appendEntries : appendEntriesRequests) { if (appendEntries.server.equals(S_03)) { assertThatAppendEntriesHasValues( appendEntries, 3, 1, 2, 1, NOOP(3, 2), NOOP(4, 3)); } else { assertThatAppendEntriesHasValues(appendEntries, 3, 1, 3, 2, NOOP(4, 3)); } } assertThatNoMoreRPCsWereSent(); verifyNoMoreInteractions(listener); assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3) ); assertThatTermAndCommitIndexHaveValues(3, 1); }
@Test public void shouldNotCommitEntriesUnlessAnEntryFromTheCurrentTermIsCommitted() throws StorageException, RPCException { insertIntoLog( NOOP(1, 1), NOOP(2, 1), NOOP(3, 2) ); store.setCommitIndex(1); becomeLeaderInTerm(3, false); Collection<AppendEntries> appendEntriesRequests; assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3) ); appendEntriesRequests = getRPCs(4, AppendEntries.class); assertThatAppendEntriesHaveValues(appendEntriesRequests, 3, 1, 3, 2, NOOP(4, 3)); assertThatNoMoreRPCsWereSent(); algorithm.onAppendEntriesReply(S_03, 3, 3, 1, false); algorithm.onAppendEntriesReply(S_01, 3, 3, 1, false); algorithm.onAppendEntriesReply(S_04, 3, 3, 1, false); algorithm.onAppendEntriesReply(S_02, 3, 3, 1, false); timer.fastForward(); appendEntriesRequests = getRPCs(4, AppendEntries.class); assertThatAppendEntriesHaveValues( appendEntriesRequests, 3, 1, 2, 1, NOOP(3, 2), NOOP(4, 3)); assertThatNoMoreRPCsWereSent(); algorithm.onAppendEntriesReply(S_03, 3, 2, 1, true); algorithm.onAppendEntriesReply(S_01, 3, 2, 1, true); algorithm.onAppendEntriesReply(S_04, 3, 2, 1, true); algorithm.onAppendEntriesReply(S_02, 3, 2, 1, true); verifyNoMoreInteractions(listener); assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3) ); assertThatTermAndCommitIndexHaveValues(3, 1); }
@Test public void shouldIgnoreDelayedAppendEntriesReplyWherePrevLogIndexGreaterThanNextIndexMinusOneAndAppliedIsFalse() throws StorageException, RPCException { insertIntoLog( NOOP(1, 1), NOOP(2, 1), NOOP(3, 2) ); becomeLeaderInTerm(3, false); assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3) ); Collection<AppendEntries> appendEntriesRequests; appendEntriesRequests = getRPCs(4, AppendEntries.class); assertThatAppendEntriesHaveValues(appendEntriesRequests, 3, 0, 3, 2, NOOP(4, 3)); assertThatNoMoreRPCsWereSent(); algorithm.onAppendEntriesReply(S_02, 3, 3, 1, false); timer.fastForward(); appendEntriesRequests = getRPCs(4, AppendEntries.class); for(AppendEntries appendEntries : appendEntriesRequests) { if (appendEntries.server.equals(S_02)) { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 2, 1, NOOP(3, 2), NOOP(4, 3)); } else { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 3, 2, NOOP(4, 3)); } } assertThatNoMoreRPCsWereSent(); algorithm.onAppendEntriesReply(S_02, 3, 2, 2, false); timer.fastForward(); appendEntriesRequests = getRPCs(4, AppendEntries.class); for(AppendEntries appendEntries : appendEntriesRequests) { if (appendEntries.server.equals(S_02)) { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 1, 1, NOOP(2, 1), NOOP(3, 2), NOOP(4, 3)); } else { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 3, 2, NOOP(4, 3)); } } assertThatNoMoreRPCsWereSent(); algorithm.onAppendEntriesReply(S_02, 3, 2, 2, false); timer.fastForward(); appendEntriesRequests = getRPCs(4, AppendEntries.class); for(AppendEntries appendEntries : appendEntriesRequests) { if (appendEntries.server.equals(S_02)) { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 1, 1, NOOP(2, 1), NOOP(3, 2), NOOP(4, 3)); } else { assertThatAppendEntriesHasValues(appendEntries, 3, 0, 3, 2, NOOP(4, 3)); } } assertThatNoMoreRPCsWereSent(); assertThatLogContains( SENTINEL(), NOOP(1, 1), NOOP(2, 1), NOOP(3, 2), NOOP(4, 3) ); assertThatTermAndCommitIndexHaveValues(3, 0); }