@Override
public boolean isInSnapshot(long logEntryIndex) {
- return logEntryIndex <= snapshotIndex && snapshotIndex != -1;
+ return logEntryIndex >= 0 && logEntryIndex <= snapshotIndex && snapshotIndex != -1;
}
@Override
followerLogInformation.setRaftVersion(appendEntriesReply.getRaftVersion());
long followerLastLogIndex = appendEntriesReply.getLogLastIndex();
- long followersLastLogTermInLeadersLog = getLogEntryTerm(followerLastLogIndex);
boolean updated = false;
if (appendEntriesReply.getLogLastIndex() > context.getReplicatedLog().lastIndex()) {
// The follower's log is actually ahead of the leader's log. Normally this doesn't happen
// However in this case the log terms won't match and the logs will conflict - this is handled
// elsewhere.
log.info("{}: handleAppendEntriesReply: follower {} lastIndex {} is ahead of our lastIndex {} "
- + "(snapshotIndex {}) - forcing install snaphot", logName(), followerLogInformation.getId(),
- appendEntriesReply.getLogLastIndex(), context.getReplicatedLog().lastIndex(),
- context.getReplicatedLog().getSnapshotIndex());
+ + "(snapshotIndex {}, snapshotTerm {}) - forcing install snaphot", logName(),
+ followerLogInformation.getId(), appendEntriesReply.getLogLastIndex(),
+ context.getReplicatedLog().lastIndex(), context.getReplicatedLog().getSnapshotIndex(),
+ context.getReplicatedLog().getSnapshotTerm());
followerLogInformation.setMatchIndex(-1);
followerLogInformation.setNextIndex(-1);
updated = true;
} else if (appendEntriesReply.isSuccess()) {
+ long followersLastLogTermInLeadersLog = getLogEntryTerm(followerLastLogIndex);
if (followerLastLogIndex >= 0 && followersLastLogTermInLeadersLog >= 0
&& followersLastLogTermInLeadersLog != appendEntriesReply.getLogLastTerm()) {
// The follower's last entry is present in the leader's journal but the terms don't match so the
updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply);
}
} else {
- log.info("{}: handleAppendEntriesReply - received unsuccessful reply: {}, leader snapshotIndex: {}",
- logName(), appendEntriesReply, context.getReplicatedLog().getSnapshotIndex());
+ log.info("{}: handleAppendEntriesReply - received unsuccessful reply: {}, leader snapshotIndex: {}, "
+ + "snapshotTerm: {}, replicatedToAllIndex: {}", logName(), appendEntriesReply,
+ context.getReplicatedLog().getSnapshotIndex(), context.getReplicatedLog().getSnapshotTerm(),
+ getReplicatedToAllIndex());
+ long followersLastLogTermInLeadersLogOrSnapshot = getLogEntryOrSnapshotTerm(followerLastLogIndex);
if (appendEntriesReply.isForceInstallSnapshot()) {
// Reset the followers match and next index. This is to signal that this follower has nothing
// in common with this Leader and so would require a snapshot to be installed
// Force initiate a snapshot capture
initiateCaptureSnapshot(followerId);
- } else if (followerLastLogIndex < 0 || followersLastLogTermInLeadersLog >= 0
- && followersLastLogTermInLeadersLog == appendEntriesReply.getLogLastTerm()) {
- // The follower's log is empty or the last entry is present in the leader's journal
- // and the terms match so the follower is just behind the leader's journal from
- // the last snapshot, if any. We'll catch up the follower quickly by starting at the
- // follower's last log index.
+ } else if (followerLastLogIndex < 0 || followersLastLogTermInLeadersLogOrSnapshot >= 0
+ && followersLastLogTermInLeadersLogOrSnapshot == appendEntriesReply.getLogLastTerm()) {
+ // The follower's log is empty or the follower's last entry is present in the leader's journal or
+ // snapshot and the terms match so the follower is just behind the leader's journal from the last
+ // snapshot, if any. We'll catch up the follower quickly by starting at the follower's last log index.
updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply);
log.info("{}: follower {} last log term {} conflicts with the leader's {} - dec next index to {}",
logName(), followerId, appendEntriesReply.getLogLastTerm(),
- followersLastLogTermInLeadersLog, followerLogInformation.getNextIndex());
+ followersLastLogTermInLeadersLogOrSnapshot, followerLogInformation.getNextIndex());
}
}
}
}
/**
- * Returns the actual term of the entry in replicated log for the given index or -1 if not found.
+ * Returns the actual term of the entry in the replicated log for the given index or -1 if not found.
*
* @return the log entry term or -1 if not found
*/
return -1;
}
+ /**
+ * Returns the actual term of the entry in the replicated log for the given index or, if not present, returns the
+ * snapshot term if the given index is in the snapshot or -1 otherwise.
+ *
+ * @return the term or -1 otherwise
+ */
+ protected long getLogEntryOrSnapshotTerm(final long index) {
+ if (context.getReplicatedLog().isInSnapshot(index)) {
+ return context.getReplicatedLog().getSnapshotTerm();
+ }
+
+ return getLogEntryTerm(index);
+ }
+
/**
* Applies the log entries up to the specified index that is known to be committed to the state machine.
*
if (lastIndex > -1) {
if (isLogEntryPresent(appendEntries.getPrevLogIndex())) {
- final long prevLogTerm = getLogEntryTerm(appendEntries.getPrevLogIndex());
- if (prevLogTerm != appendEntries.getPrevLogTerm()) {
+ final long leadersPrevLogTermInFollowersLogOrSnapshot =
+ getLogEntryOrSnapshotTerm(appendEntries.getPrevLogIndex());
+ if (leadersPrevLogTermInFollowersLogOrSnapshot != appendEntries.getPrevLogTerm()) {
// The follower's log is out of sync because the Leader's prevLogIndex entry does exist
- // in the follower's log but it has a different term in it
+ // in the follower's log or snapshot but it has a different term.
log.info("{}: The prevLogIndex {} was found in the log but the term {} is not equal to the append "
- + "entries prevLogTerm {} - lastIndex: {}, snapshotIndex: {}", logName(),
- appendEntries.getPrevLogIndex(), prevLogTerm, appendEntries.getPrevLogTerm(), lastIndex,
- context.getReplicatedLog().getSnapshotIndex());
+ + "entries prevLogTerm {} - lastIndex: {}, snapshotIndex: {}, snapshotTerm: {}", logName(),
+ appendEntries.getPrevLogIndex(), leadersPrevLogTermInFollowersLogOrSnapshot,
+ appendEntries.getPrevLogTerm(), lastIndex, context.getReplicatedLog().getSnapshotIndex(),
+ context.getReplicatedLog().getSnapshotTerm());
sendOutOfSyncAppendEntriesReply(sender, false);
return true;
// The follower's log is out of sync because the Leader's prevLogIndex entry was not found in it's log
log.info("{}: The log is not empty but the prevLogIndex {} was not found in it - lastIndex: {}, "
- + "snapshotIndex: {}", logName(), appendEntries.getPrevLogIndex(), lastIndex,
- context.getReplicatedLog().getSnapshotIndex());
+ + "snapshotIndex: {}, snapshotTerm: {}", logName(), appendEntries.getPrevLogIndex(), lastIndex,
+ context.getReplicatedLog().getSnapshotIndex(), context.getReplicatedLog().getSnapshotTerm());
sendOutOfSyncAppendEntriesReply(sender, false);
return true;
// the previous entry in it's in-memory journal
log.info("{}: Cannot append entries because the replicatedToAllIndex {} does not appear to be in the "
- + "in-memory journal", logName(), appendEntries.getReplicatedToAllIndex());
+ + "in-memory journal - lastIndex: {}, snapshotIndex: {}, snapshotTerm: {}", logName(),
+ appendEntries.getReplicatedToAllIndex(), lastIndex,
+ context.getReplicatedLog().getSnapshotIndex(), context.getReplicatedLog().getSnapshotTerm());
sendOutOfSyncAppendEntriesReply(sender, false);
return true;
final List<ReplicatedLogEntry> entries = appendEntries.getEntries();
if (entries.size() > 0 && !isLogEntryPresent(entries.get(0).getIndex() - 1)) {
log.info("{}: Cannot append entries because the calculated previousIndex {} was not found in the "
- + "in-memory journal", logName(), entries.get(0).getIndex() - 1);
+ + "in-memory journal - lastIndex: {}, snapshotIndex: {}, snapshotTerm: {}", logName(),
+ entries.get(0).getIndex() - 1, lastIndex, context.getReplicatedLog().getSnapshotIndex(),
+ context.getReplicatedLog().getSnapshotTerm());
sendOutOfSyncAppendEntriesReply(sender, false);
return true;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
-import static org.mockito.Matchers.any;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
@Test
public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog() {
- logStart("testHandleFirstAppendEntries");
+ logStart("testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog");
MockRaftActorContext context = createActorContext();
context.getReplicatedLog().clear(0,2);
@Test
public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot() {
- logStart("testHandleFirstAppendEntries");
+ logStart("testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot");
MockRaftActorContext context = createActorContext();
context.getReplicatedLog().clear(0,2);
}
/**
- * This test verifies that when an AppendEntries is received a specific prevLogTerm
+ * This test verifies that when an AppendEntries is received with a prevLogTerm
* which does not match the term that is in RaftActors log entry at prevLogIndex
* then the RaftActor does not change it's state and it returns a failure.
*/
MockRaftActorContext context = createActorContext();
- // First set the receivers term to lower number
- context.getTermInformation().update(95, "test");
-
- // AppendEntries is now sent with a bigger term
- // this will set the receivers term to be the same as the sender's term
- AppendEntries appendEntries = new AppendEntries(100, "leader", 0, 0, Collections.emptyList(), 101, -1,
- (short)0);
+ AppendEntries appendEntries = new AppendEntries(2, "leader", 0, 2, Collections.emptyList(), 101, -1, (short)0);
follower = createBehavior(context);
assertEquals("isSuccess", false, reply.isSuccess());
}
+ @Test
+ public void testHandleAppendEntriesSenderPrevLogIndexIsInTheSnapshot() {
+ logStart("testHandleAppendEntriesSenderPrevLogIndexIsInTheSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+ context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(5, 8, 3).build());
+ context.getReplicatedLog().setSnapshotIndex(4);
+ context.getReplicatedLog().setSnapshotTerm(3);
+
+ AppendEntries appendEntries = new AppendEntries(3, "leader", 1, 3, Collections.emptyList(), 8, -1, (short)0);
+
+ follower = createBehavior(context);
+
+ RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
+
+ Assert.assertSame(follower, newBehavior);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+
+ assertEquals("isSuccess", true, reply.isSuccess());
+ }
+
/**
* This test verifies that when a new AppendEntries message is received with
* new entries and the logs of the sender and receiver match that the new