package org.opendaylight.controller.cluster.raft;
import static org.junit.Assert.assertEquals;
+
import akka.persistence.SaveSnapshotSuccess;
import com.google.common.collect.ImmutableMap;
import java.util.List;
import org.junit.Test;
import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
-import org.opendaylight.controller.cluster.raft.RaftActor.UpdateElectionTerm;
-import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
-import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
+import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
private MockPayload recoveredPayload0;
private MockPayload recoveredPayload1;
private MockPayload recoveredPayload2;
+ private MockPayload payload3;
private MockPayload payload4;
private MockPayload payload5;
private MockPayload payload6;
private MockPayload payload7;
@Test
- public void runTest() {
+ public void runTest() throws Exception {
testLog.info("testReplicationAndSnapshots starting");
// Setup the persistent journal for the leader. We'll start up with 3 journal log entries (one less
InMemoryJournal.addEntry(leaderId, seqId++, new ReplicatedLogImplEntry(1, initialTerm, recoveredPayload1));
recoveredPayload2 = new MockPayload("two");
InMemoryJournal.addEntry(leaderId, seqId++, new ReplicatedLogImplEntry(2, initialTerm, recoveredPayload2));
- InMemoryJournal.addEntry(leaderId, seqId++, new ApplyLogEntries(2));
+ InMemoryJournal.addEntry(leaderId, seqId++, new ApplyJournalEntries(2));
origLeaderJournal = InMemoryJournal.get(leaderId, ReplicatedLogImplEntry.class);
// Create the leader and 2 follower actors and verify initial syncing of the followers after leader
// persistence recovery.
+ DefaultConfigParamsImpl followerConfigParams = newFollowerConfigParams();
+ followerConfigParams.setSnapshotBatchCount(snapshotBatchCount);
follower1Actor = newTestRaftActor(follower1Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
- follower2Id, testActorPath(follower2Id)), newFollowerConfigParams());
+ follower2Id, testActorPath(follower2Id)), followerConfigParams);
follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
- follower1Id, testActorPath(follower1Id)), newFollowerConfigParams());
+ follower1Id, testActorPath(follower1Id)), followerConfigParams);
- peerAddresses = ImmutableMap.<String, String>builder().
- put(follower1Id, follower1Actor.path().toString()).
- put(follower2Id, follower2Actor.path().toString()).build();
+ peerAddresses = ImmutableMap.<String, String>builder()
+ .put(follower1Id, follower1Actor.path().toString())
+ .put(follower2Id, follower2Actor.path().toString()).build();
leaderConfigParams = newLeaderConfigParams();
leaderActor = newTestRaftActor(leaderId, peerAddresses, leaderConfigParams);
* scenario, the follower consensus and application of state is delayed until after the snapshot
* completes.
*/
- private void testFirstSnapshot() {
+ private void testFirstSnapshot() throws Exception {
testLog.info("testFirstSnapshot starting");
- byte[] snapshot = new byte[] {1,2,3,4};
- leaderActor.underlyingActor().setSnapshot(snapshot);
+ expSnapshotState.add(recoveredPayload0);
+ expSnapshotState.add(recoveredPayload1);
+ expSnapshotState.add(recoveredPayload2);
// Delay the consensus by temporarily dropping the AppendEntries to both followers.
follower1Actor.underlyingActor().startDropMessages(AppendEntries.class);
follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
// Send the payload.
- MockPayload payload3 = sendPayloadData(leaderActor, "three");
+ payload3 = sendPayloadData(leaderActor, "three");
// Wait for snapshot complete.
MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
// the last applied log entry (2) even though the leader hasn't yet advanced its cached snapshot index.
List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
- verifySnapshot("Persisted", persistedSnapshots.get(0), initialTerm, 2, currentTerm, 3, snapshot);
+ verifySnapshot("Persisted", persistedSnapshots.get(0), initialTerm, 2, currentTerm, 3);
List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 3, payload3);
// The leader's persisted journal log should be cleared since we snapshotted.
- List<ReplicatedLogImplEntry> persistedLeaderJournal = InMemoryJournal.get(leaderId, ReplicatedLogImplEntry.class);
+ List<ReplicatedLogImplEntry> persistedLeaderJournal =
+ InMemoryJournal.get(leaderId, ReplicatedLogImplEntry.class);
assertEquals("Persisted journal log size", 0, persistedLeaderJournal.size());
// Allow AppendEntries to both followers to proceed. This should catch up the followers and cause a
assertEquals("Leader last applied", 3, leaderContext.getLastApplied());
assertEquals("Leader replicatedToAllIndex", 2, leader.getReplicatedToAllIndex());
+ // The followers should also snapshot so verify.
+
+ MessageCollectorActor.expectFirstMatching(follower1CollectorActor, SaveSnapshotSuccess.class);
+ persistedSnapshots = InMemorySnapshotStore.getSnapshots(follower1Id, Snapshot.class);
+ assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
+ // The last applied index in the snapshot may or may not be the last log entry depending on
+ // timing so to avoid intermittent test failures, we'll just verify the snapshot's last term/index.
+ assertEquals("Follower1 Snapshot getLastTerm", currentTerm, persistedSnapshots.get(0).getLastTerm());
+ assertEquals("Follower1 Snapshot getLastIndex", 3, persistedSnapshots.get(0).getLastIndex());
+
+ MessageCollectorActor.expectFirstMatching(follower2CollectorActor, SaveSnapshotSuccess.class);
+
MessageCollectorActor.clearMessages(leaderCollectorActor);
MessageCollectorActor.clearMessages(follower1CollectorActor);
MessageCollectorActor.clearMessages(follower2CollectorActor);
* Send one more payload to trigger another snapshot. In this scenario, we delay the snapshot until
* consensus occurs and the leader applies the state.
*/
- private void testSecondSnapshot() {
+ private void testSecondSnapshot() throws Exception {
testLog.info("testSecondSnapshot starting");
- byte[] snapshot = new byte[] {5,6,7,8};
- leaderActor.underlyingActor().setSnapshot(snapshot);
+ expSnapshotState.add(payload3);
+ expSnapshotState.add(payload4);
+ expSnapshotState.add(payload5);
+ expSnapshotState.add(payload6);
// Delay the CaptureSnapshot message to the leader actor.
- leaderActor.underlyingActor().startDropMessages(CaptureSnapshot.class);
+ leaderActor.underlyingActor().startDropMessages(CaptureSnapshotReply.class);
// Send the payload.
payload7 = sendPayloadData(leaderActor, "seven");
- // Capture the CaptureSnapshot message so we can send it later.
- CaptureSnapshot captureSnapshot = MessageCollectorActor.expectFirstMatching(
- leaderCollectorActor, CaptureSnapshot.class);
+ // Capture the CaptureSnapshotReply message so we can send it later.
+ final CaptureSnapshotReply captureSnapshotReply = MessageCollectorActor.expectFirstMatching(
+ leaderCollectorActor, CaptureSnapshotReply.class);
// Wait for the state to be applied in the leader.
ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
assertEquals("Leader last applied", 7, leaderContext.getLastApplied());
assertEquals("Leader replicatedToAllIndex", 5, leader.getReplicatedToAllIndex());
- // Now deliver the CaptureSnapshot.
- leaderActor.underlyingActor().stopDropMessages(CaptureSnapshot.class);
- leaderActor.tell(captureSnapshot, leaderActor);
-
- // Wait for CaptureSnapshotReply to complete.
- MessageCollectorActor.expectFirstMatching(leaderCollectorActor, CaptureSnapshotReply.class);
+ // Now deliver the CaptureSnapshotReply.
+ leaderActor.underlyingActor().stopDropMessages(CaptureSnapshotReply.class);
+ leaderActor.tell(captureSnapshotReply, leaderActor);
// Wait for snapshot complete.
MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
assertEquals("Leader journal last index", 7, leaderContext.getReplicatedLog().lastIndex());
assertEquals("Leader commit index", 7, leaderContext.getCommitIndex());
- // Verify the persisted snapshot. This should reflect the advanced snapshot index as the last applied
- // log entry (6).
+ // Verify the persisted snapshot. This should reflect the snapshot index as the last applied
+ // log entry (7) and shouldn't contain any unapplied entries as we capture persisted the snapshot data
+ // when the snapshot is created (ie when the CaptureSnapshot is processed).
List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
- verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 6, currentTerm, 7, snapshot);
+ verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 6, currentTerm, 7);
List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 7, payload7);
assertEquals("Persisted journal log size", 0, persistedLeaderJournal.size());
// Verify the followers apply all 4 new log entries.
- List<ApplyState> applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 4);
+ List<ApplyState> applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor,
+ ApplyState.class, 4);
verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
MessageCollectorActor.clearMessages(follower1CollectorActor);
MessageCollectorActor.expectFirstMatching(follower1CollectorActor, AppendEntries.class);
- RaftActorContext follower1Context = follower1Actor.underlyingActor().getRaftActorContext();
+ follower1Context = follower1Actor.underlyingActor().getRaftActorContext();
assertEquals("Follower 1 snapshot term", currentTerm, follower1Context.getReplicatedLog().getSnapshotTerm());
assertEquals("Follower 1 snapshot index", 6, follower1Context.getReplicatedLog().getSnapshotIndex());
assertEquals("Follower 1 journal log size", 1, follower1Context.getReplicatedLog().size());
MessageCollectorActor.clearMessages(follower2CollectorActor);
MessageCollectorActor.expectFirstMatching(follower2CollectorActor, AppendEntries.class);
- RaftActorContext follower2Context = follower2Actor.underlyingActor().getRaftActorContext();
+ follower2Context = follower2Actor.underlyingActor().getRaftActorContext();
assertEquals("Follower 2 snapshot term", currentTerm, follower2Context.getReplicatedLog().getSnapshotTerm());
assertEquals("Follower 2 snapshot index", 6, follower2Context.getReplicatedLog().getSnapshotIndex());
assertEquals("Follower 2 journal log size", 1, follower2Context.getReplicatedLog().size());
assertEquals("Follower 2 journal last index", 7, follower2Context.getReplicatedLog().lastIndex());
assertEquals("Follower 2 commit index", 7, follower2Context.getCommitIndex());
+ expSnapshotState.add(payload7);
+
testLog.info("testSecondSnapshot ending");
}
leaderActor.underlyingActor().waitForRecoveryComplete();
+ leaderContext = leaderActor.underlyingActor().getRaftActorContext();
+
assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
assertEquals("Leader snapshot index", 6, leaderContext.getReplicatedLog().getSnapshotIndex());
assertEquals("Leader journal log size", 1, leaderContext.getReplicatedLog().size());
Code Review / controller.git / blobdiff