@Test public void shouldReturnLogEntriesInResponseToRepeatedGetNextCommittedCallsWithInitialIndexToSearchFromNonZeroAndOverlapBetweenLogAndSnapshot() throws Exception { long lastAppliedIndex = 9; SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(lastAppliedIndex, 2); LogEntry unappliedEntry0 = CLIENT(11, 3, new UnitTestCommand()); final LogEntry[] entries = new LogEntry[] { CLIENT(7, 2, new UnitTestCommand()), NOOP(8, 2), NOOP(9, 3), CLIENT(10, 3, new UnitTestCommand()), unappliedEntry0, CLIENT(12, 3, new UnitTestCommand()) }; clearLog(); insertIntoLog(entries); long currentTerm = 3; store.setCurrentTerm(currentTerm); long commitIndex = 11; store.setCommitIndex(commitIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); Committed committed; committed = algorithm.getNextCommitted(10); checkMatchingCommittedLogEntry(committed, unappliedEntry0); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Override public synchronized @Nullable Committed getNextCommitted(long indexToSearchFrom) { LOGGER.trace("{}: get next committed from {}", self, indexToSearchFrom); checkState(initialized); checkArgument(indexToSearchFrom >= 0, "index:%s must be positive", indexToSearchFrom); try { long currentTerm = store.getCurrentTerm(); long commitIndex = store.getCommitIndex(); ExtendedSnapshot latestSnapshot = snapshotsStore.getLatestSnapshot(); checkSnapshotLogAndCommitIndicesAndTerms(latestSnapshot, log.getFirst(), log.getLast(), currentTerm, commitIndex); checkArgument(indexToSearchFrom <= commitIndex, "indexToSearchFrom:%s commitIndex:%s", indexToSearchFrom, commitIndex); if (indexToSearchFrom == commitIndex) { return null; } Committed committed = null; for (long index = indexToSearchFrom + 1; index <= commitIndex; index++) { LogEntry currentLog = log.get(index); if (currentLog == null) { break; } committed = convertLogEntryToCommitted(currentLog); if (committed != null) { break; } } if ((committed == null) && (latestSnapshot != null) && (indexToSearchFrom < latestSnapshot.getIndex())) { committed = latestSnapshot; } return committed; } catch (StorageException e) { handleStorageException(e); throw new RaftError("handleStorageException did not throw", e); } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized @Nullable Committed getNextCommitted(long indexToSearchFrom) { LOGGER.trace("{}: get next committed from {}", self, indexToSearchFrom); checkState(initialized); checkArgument(indexToSearchFrom >= 0, "index:%s must be positive", indexToSearchFrom); try { long currentTerm = store.getCurrentTerm(); long commitIndex = store.getCommitIndex(); ExtendedSnapshot latestSnapshot = snapshotsStore.getLatestSnapshot(); checkSnapshotLogAndCommitIndicesAndTerms(latestSnapshot, log.getFirst(), log.getLast(), currentTerm, commitIndex); checkArgument(indexToSearchFrom <= commitIndex, "indexToSearchFrom:%s commitIndex:%s", indexToSearchFrom, commitIndex); if (indexToSearchFrom == commitIndex) { return null; } Committed committed = null; for (long index = indexToSearchFrom + 1; index <= commitIndex; index++) { LogEntry currentLog = log.get(index); if (currentLog == null) { break; } committed = convertLogEntryToCommitted(currentLog); if (committed != null) { break; } } if ((committed == null) && (latestSnapshot != null) && (indexToSearchFrom < latestSnapshot.getIndex())) { committed = latestSnapshot; } return committed; } catch (StorageException e) { handleStorageException(e); throw new RaftError("handleStorageException did not throw", e); } } }
RaftAlgorithm implements RPCReceiver, Raft { @Override public synchronized @Nullable Committed getNextCommitted(long indexToSearchFrom) { LOGGER.trace("{}: get next committed from {}", self, indexToSearchFrom); checkState(initialized); checkArgument(indexToSearchFrom >= 0, "index:%s must be positive", indexToSearchFrom); try { long currentTerm = store.getCurrentTerm(); long commitIndex = store.getCommitIndex(); ExtendedSnapshot latestSnapshot = snapshotsStore.getLatestSnapshot(); checkSnapshotLogAndCommitIndicesAndTerms(latestSnapshot, log.getFirst(), log.getLast(), currentTerm, commitIndex); checkArgument(indexToSearchFrom <= commitIndex, "indexToSearchFrom:%s commitIndex:%s", indexToSearchFrom, commitIndex); if (indexToSearchFrom == commitIndex) { return null; } Committed committed = null; for (long index = indexToSearchFrom + 1; index <= commitIndex; index++) { LogEntry currentLog = log.get(index); if (currentLog == null) { break; } committed = convertLogEntryToCommitted(currentLog); if (committed != null) { break; } } if ((committed == null) && (latestSnapshot != null) && (indexToSearchFrom < latestSnapshot.getIndex())) { committed = latestSnapshot; } return committed; } catch (StorageException e) { handleStorageException(e); throw new RaftError("handleStorageException did not throw", 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 @Nullable Committed getNextCommitted(long indexToSearchFrom) { LOGGER.trace("{}: get next committed from {}", self, indexToSearchFrom); checkState(initialized); checkArgument(indexToSearchFrom >= 0, "index:%s must be positive", indexToSearchFrom); try { long currentTerm = store.getCurrentTerm(); long commitIndex = store.getCommitIndex(); ExtendedSnapshot latestSnapshot = snapshotsStore.getLatestSnapshot(); checkSnapshotLogAndCommitIndicesAndTerms(latestSnapshot, log.getFirst(), log.getLast(), currentTerm, commitIndex); checkArgument(indexToSearchFrom <= commitIndex, "indexToSearchFrom:%s commitIndex:%s", indexToSearchFrom, commitIndex); if (indexToSearchFrom == commitIndex) { return null; } Committed committed = null; for (long index = indexToSearchFrom + 1; index <= commitIndex; index++) { LogEntry currentLog = log.get(index); if (currentLog == null) { break; } committed = convertLogEntryToCommitted(currentLog); if (committed != null) { break; } } if ((committed == null) && (latestSnapshot != null) && (indexToSearchFrom < latestSnapshot.getIndex())) { committed = latestSnapshot; } return committed; } catch (StorageException e) { handleStorageException(e); throw new RaftError("handleStorageException did not throw", 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 @Nullable Committed getNextCommitted(long indexToSearchFrom) { LOGGER.trace("{}: get next committed from {}", self, indexToSearchFrom); checkState(initialized); checkArgument(indexToSearchFrom >= 0, "index:%s must be positive", indexToSearchFrom); try { long currentTerm = store.getCurrentTerm(); long commitIndex = store.getCommitIndex(); ExtendedSnapshot latestSnapshot = snapshotsStore.getLatestSnapshot(); checkSnapshotLogAndCommitIndicesAndTerms(latestSnapshot, log.getFirst(), log.getLast(), currentTerm, commitIndex); checkArgument(indexToSearchFrom <= commitIndex, "indexToSearchFrom:%s commitIndex:%s", indexToSearchFrom, commitIndex); if (indexToSearchFrom == commitIndex) { return null; } Committed committed = null; for (long index = indexToSearchFrom + 1; index <= commitIndex; index++) { LogEntry currentLog = log.get(index); if (currentLog == null) { break; } committed = convertLogEntryToCommitted(currentLog); if (committed != null) { break; } } if ((committed == null) && (latestSnapshot != null) && (indexToSearchFrom < latestSnapshot.getIndex())) { committed = latestSnapshot; } return committed; } catch (StorageException e) { handleStorageException(e); throw new RaftError("handleStorageException did not throw", 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);  }