X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FReplicationAndSnapshotsIntegrationTest.java;h=c658ef08045ee275878e8c16ab1575fdeeeda3db;hp=bd670fd581153e673ef145ca178211c6f7e60f86;hb=9d5ec5cdd146a56bc03e35b6718e9492a5c8410a;hpb=3940ce6060e027fe870244346e5309229cc8dc48 diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsIntegrationTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsIntegrationTest.java index bd670fd581..c658ef0804 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsIntegrationTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsIntegrationTest.java @@ -13,11 +13,10 @@ 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.ApplyJournalEntries; 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.base.messages.UpdateElectionTerm; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; @@ -36,13 +35,14 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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 @@ -55,18 +55,20 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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.builder(). put(follower1Id, follower1Actor.path().toString()). @@ -157,19 +159,21 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt * 4 and we already have 3 entries in the journal log, this should initiate a snapshot. In this * scenario, the follower consensus and application of state is delayed until after the snapshot * completes. + * @throws Exception */ - 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); @@ -185,7 +189,7 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt // the last applied log entry (2) even though the leader hasn't yet advanced its cached snapshot index. List 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 unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries(); assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size()); verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 3, payload3); @@ -224,6 +228,18 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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); @@ -286,22 +302,25 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt /** * Send one more payload to trigger another snapshot. In this scenario, we delay the snapshot until * consensus occurs and the leader applies the state. + * @throws Exception */ - 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. + CaptureSnapshotReply captureSnapshotReply = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, + CaptureSnapshotReply.class); // Wait for the state to be applied in the leader. ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class); @@ -321,12 +340,9 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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); @@ -341,11 +357,12 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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 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 unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries(); assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size()); verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 7, payload7); @@ -389,6 +406,8 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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"); } @@ -404,6 +423,8 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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());