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);  }
@Test public void shouldThrowIllegalArgumentExceptionIfCallerSpecifiesIndexLessThanZero() { expectedException.expect(IllegalArgumentException.class); algorithm.getNextCommitted(-1); }
@Test public void shouldThrowIllegalArgumentExceptionIfCallerSpecifiesIndexGreaterThanCommitIndexInCallToGetNextCommitted() throws Exception { final LogEntry[] entries = new LogEntry[] { SENTINEL(), NOOP(1, 1), CLIENT(2, 1, new UnitTestCommand()), CLIENT(3, 1, new UnitTestCommand()), CLIENT(4, 1, new UnitTestCommand()), CLIENT(5, 1, new UnitTestCommand()), CLIENT(6, 1, new UnitTestCommand()) }; final long currentTerm = 1; final long commitIndex = 3; insertIntoLog(entries); store.setCurrentTerm(currentTerm); store.setCommitIndex(commitIndex); IllegalArgumentException callException = null; try { long callerLastAppliedCommandIndex = commitIndex + 1; assertThat(callerLastAppliedCommandIndex, lessThanOrEqualTo((long) entries.length - 1)); algorithm.getNextCommitted(callerLastAppliedCommandIndex); } catch (IllegalArgumentException e) { callException = e; } assertThat(callException, notNullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Test public void shouldThrowIllegalArgumentExceptionIfCallerSpecifiesIndexGreaterThanLastLogIndexInCallToGetNextCommitted() throws Exception { final LogEntry[] entries = new LogEntry[] { SENTINEL(), NOOP(1, 1), CLIENT(2, 1, new UnitTestCommand()), CLIENT(3, 1, new UnitTestCommand()), CLIENT(4, 1, new UnitTestCommand()), CLIENT(5, 1, new UnitTestCommand()), CLIENT(6, 1, new UnitTestCommand()) }; final long currentTerm = 2; final long commitIndex = entries.length - 1; insertIntoLog(entries); store.setCurrentTerm(currentTerm); store.setCommitIndex(commitIndex); IllegalArgumentException callException = null; try { algorithm.getNextCommitted(entries.length); } catch (IllegalArgumentException e) { callException = e; } assertThat(callException, notNullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Test public void shouldThrowIllegalArgumentExceptionIfCallerSpecifiesIndexGreaterThanLastAppliedIndexInSnapshotInCallToGetNextCommitted() throws Exception { long lastAppliedIndex = 6; SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(lastAppliedIndex, 2); clearLog(); long currentTerm = 2; store.setCurrentTerm(currentTerm); store.setCommitIndex(lastAppliedIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); IllegalArgumentException callException = null; try { algorithm.getNextCommitted(lastAppliedIndex + 1); } catch (IllegalArgumentException e) { callException = e; } assertThat(callException, notNullValue()); assertThatLogIsEmpty(); assertThatTermAndCommitIndexHaveValues(currentTerm, lastAppliedIndex); }
@Test public void shouldThrowIllegalArgumentExceptionIfCallerSpecifiesIndexGreaterThanLastLogIndexInCallToGetNextCommittedAndBothLogAndSnapshotExist() throws Exception { SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(9L, 2); LogEntry unappliedEntry0 = CLIENT(7, 2, new UnitTestCommand()); LogEntry unAppliedEntry1 = NOOP(8, 2); LogEntry unappliedEntry2 = NOOP(9, 3); LogEntry unappliedEntry3 = CLIENT(10, 3, new UnitTestCommand()); LogEntry unappliedEntry4 = CLIENT(11, 3, new UnitTestCommand()); final LogEntry[] entries = new LogEntry[] { unappliedEntry0, unAppliedEntry1, unappliedEntry2, unappliedEntry3, unappliedEntry4, 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); IllegalArgumentException callException = null; try { algorithm.getNextCommitted(entries[entries.length - 1].getIndex() + 1); } catch (IllegalArgumentException e) { callException = e; } assertThat(callException, notNullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Test public void shouldReturnNoUnappliedEntriesIfTheLogIsEmptyAndThereAreNoSnapshots() throws Exception { insertIntoLog(SENTINEL()); store.setCommitIndex(0); when(snapshotsStore.getLatestSnapshot()).thenReturn(null); assertThat(algorithm.getNextCommitted(0), nullValue()); assertThatLogContainsOnlySentinel(); assertThatTermAndCommitIndexHaveValues(0, 0); }
@Test public void shouldReturnSnapshotInResponseToRepeatedGetNextCommittedCallsWithInitialIndexToSearchFromAsZero() throws Exception { int lastAppliedTerm = 2; long lastAppliedIndex = 6; SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(lastAppliedIndex, lastAppliedTerm); clearLog(); store.setCommitIndex(lastAppliedIndex); store.setCurrentTerm(lastAppliedTerm); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); Committed committed; committed = algorithm.getNextCommitted(0); checkMatchingSnapshot(committed, storedSnapshot); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogIsEmpty(); assertThatTermAndCommitIndexHaveValues(lastAppliedTerm, lastAppliedIndex); }
@Test public void shouldReturnSnapshotInResponseToRepeatedGetNextCommittedCallsWithInitialIndexToSearchFromIsBeforeLastAppliedIndexInSnapshot() throws Exception { long lastAppliedTerm = 2; long lastAppliedIndex = 6; SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(lastAppliedIndex, lastAppliedTerm); clearLog(); long currentTerm = lastAppliedTerm + 1; store.setCurrentTerm(currentTerm); store.setCommitIndex(lastAppliedIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); Committed committed; committed = algorithm.getNextCommitted(3); checkMatchingSnapshot(committed, storedSnapshot); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogIsEmpty(); assertThatTermAndCommitIndexHaveValues(currentTerm, lastAppliedIndex); }
@Test public void shouldReturnSnapshotFollowedByLogEntriesInResponseToRepeatedGetNextCommittedCallsWithInitialIndexToSearchFromAsZero() throws Exception { SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot((long) 6, 2); LogEntry unappliedEntry0 = CLIENT(7, 2, new UnitTestCommand()); LogEntry unappliedEntry1 = NOOP(8, 2); final LogEntry[] entries = new LogEntry[] { unappliedEntry0, unappliedEntry1, NOOP(9, 3), CLIENT(10, 3, new UnitTestCommand()), CLIENT(11, 3, new UnitTestCommand()), CLIENT(12, 3, new UnitTestCommand()) }; clearLog(); insertIntoLog(entries); long currentTerm = 8; store.setCurrentTerm(currentTerm); long commitIndex = 8; store.setCommitIndex(commitIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); Committed committed; committed = algorithm.getNextCommitted(0); checkMatchingSnapshot(committed, storedSnapshot); committed = algorithm.getNextCommitted(committed.getIndex()); checkMatchingCommittedLogEntry(committed, unappliedEntry0); committed = algorithm.getNextCommitted(committed.getIndex()); checkMatchingCommittedLogEntry(committed, unappliedEntry1); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Test public void shouldReturnLogEntriesInResponseToRepeatedGetNextCommittedCallsWithInitialIndexToSearchFromNonZeroAndAfterTheLastAppliedIndexInTheSnapshot() throws Exception { SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(6L, 2); LogEntry unappliedEntry0 = NOOP(8, 2); LogEntry unappliedEntry1 = CLIENT(9, 2, new UnitTestCommand()); final LogEntry[] entries = new LogEntry[] { CLIENT(7, 2, new UnitTestCommand()), unappliedEntry0, unappliedEntry1, CLIENT(10, 3, new UnitTestCommand()), CLIENT(11, 3, new UnitTestCommand()), CLIENT(12, 3, new UnitTestCommand()) }; clearLog(); insertIntoLog(entries); long currentTerm = 3; store.setCurrentTerm(currentTerm); long commitIndex = 9; store.setCommitIndex(commitIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); Committed committed; committed = algorithm.getNextCommitted(7); checkMatchingCommittedLogEntry(committed, unappliedEntry0); committed = algorithm.getNextCommitted(committed.getIndex()); checkMatchingCommittedLogEntry(committed, unappliedEntry1); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Test public void shouldReturnSnapshotFollowedByLogEntriesInResponseToRepeatedGetNextCommittedCallsWithInitialIndexToSearchFromAsZeroAndOverlapBetweenLogAndSnapshot() throws Exception { SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(9L, 2); LogEntry unappliedEntries[] = { CLIENT(7, 2, new UnitTestCommand()), NOOP(8, 2), NOOP(9, 3), CLIENT(10, 3, new UnitTestCommand()), CLIENT(11, 3, new UnitTestCommand()), }; final LogEntry[] entries = new LogEntry[] { unappliedEntries[0], unappliedEntries[1], unappliedEntries[2], unappliedEntries[3], unappliedEntries[4], CLIENT(12, 3, new UnitTestCommand()) }; clearLog(); insertIntoLog(entries); long currentTerm = 5; store.setCurrentTerm(currentTerm); long commitIndex = 11; store.setCommitIndex(commitIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); Committed committed; committed = algorithm.getNextCommitted(0); checkMatchingSnapshot(committed, storedSnapshot); committed = algorithm.getNextCommitted(committed.getIndex()); checkMatchingCommittedLogEntry(committed, unappliedEntries[3]); committed = algorithm.getNextCommitted(committed.getIndex()); checkMatchingCommittedLogEntry(committed, unappliedEntries[4]); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Test public void shouldReturnSnapshotFollowedByLogEntriesInResponseToRepeatedGetNextCommittedCallsWithInitialIndexToSearchFromNonZeroAndOverlapBetweenLogAndSnapshot() throws Exception { SnapshotsStore.ExtendedSnapshot storedSnapshot = new UnitTestSnapshot(9L, 2); LogEntry unappliedEntries[] = { CLIENT(7, 2, new UnitTestCommand()), NOOP(8, 2), NOOP(9, 3), CLIENT(10, 3, new UnitTestCommand()), CLIENT(11, 3, new UnitTestCommand()), }; final LogEntry[] entries = new LogEntry[] { unappliedEntries[0], unappliedEntries[1], unappliedEntries[2], unappliedEntries[3], unappliedEntries[4], CLIENT(12, 3, new UnitTestCommand()) }; clearLog(); insertIntoLog(entries); long currentTerm = 4; store.setCurrentTerm(currentTerm); long commitIndex = 11; store.setCommitIndex(commitIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(storedSnapshot); Committed committed; committed = algorithm.getNextCommitted(5); checkMatchingSnapshot(committed, storedSnapshot); committed = algorithm.getNextCommitted(committed.getIndex()); checkMatchingCommittedLogEntry(committed, unappliedEntries[3]); committed = algorithm.getNextCommitted(committed.getIndex()); checkMatchingCommittedLogEntry(committed, unappliedEntries[4]); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@Test public void shouldReturnLogEntryInResponseToRepeatedGetNextCommittedCallsWithNonZeroInitialIndexToSearchFromAndOverlapBetweenLogAndSnapshot() throws Exception { SnapshotsStore.ExtendedSnapshot snapshot = new UnitTestSnapshot(7L, 2); LogEntry unappliedEntry0 = NOOP(7, 2); final LogEntry[] entries = new LogEntry[] { unappliedEntry0, NOOP(8, 2), NOOP(9, 3), CLIENT(10, 3, new UnitTestCommand()), CLIENT(11, 3, new UnitTestCommand()), CLIENT(12, 3, new UnitTestCommand()) }; clearLog(); insertIntoLog(entries); long currentTerm = 3; store.setCurrentTerm(currentTerm); long commitIndex = 7; store.setCommitIndex(commitIndex); when(snapshotsStore.getLatestSnapshot()).thenReturn(snapshot); Committed committed; committed = algorithm.getNextCommitted(6); checkMatchingCommittedLogEntry(committed, unappliedEntry0); committed = algorithm.getNextCommitted(committed.getIndex()); assertThat(committed, nullValue()); assertThatLogContains(entries); assertThatTermAndCommitIndexHaveValues(currentTerm, commitIndex); }
@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); }