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=386108f61529258a13fb91c5f83eac8c41c4d8f3;hp=c74705d13f6c682f30d2553665a34552f62bbc47;hb=e1eca73a5ae2ffae8dd78c6fe5281cd2f45d5ef3;hpb=c190bfc14468c9ad954201a53326df941161c470 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 c74705d13f..386108f615 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 @@ -8,18 +8,18 @@ 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.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.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; @@ -63,15 +63,17 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt // 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()). - put(follower2Id, follower2Actor.path().toString()).build(); + peerAddresses = ImmutableMap.builder() + .put(follower1Id, follower1Actor.path().toString()) + .put(follower2Id, follower2Actor.path().toString()).build(); leaderConfigParams = newLeaderConfigParams(); leaderActor = newTestRaftActor(leaderId, peerAddresses, leaderConfigParams); @@ -158,7 +160,6 @@ 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() throws Exception { testLog.info("testFirstSnapshot starting"); @@ -194,7 +195,8 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 3, payload3); // The leader's persisted journal log should be cleared since we snapshotted. - List persistedLeaderJournal = InMemoryJournal.get(leaderId, ReplicatedLogImplEntry.class); + List 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 @@ -227,6 +229,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); @@ -289,7 +303,6 @@ 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() throws Exception { testLog.info("testSecondSnapshot starting"); @@ -300,14 +313,14 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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); @@ -327,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); @@ -347,8 +357,6 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt assertEquals("Leader journal last index", 7, leaderContext.getReplicatedLog().lastIndex()); assertEquals("Leader commit index", 7, leaderContext.getCommitIndex()); - expSnapshotState.add(payload7); - // 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). @@ -365,7 +373,8 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt assertEquals("Persisted journal log size", 0, persistedLeaderJournal.size()); // Verify the followers apply all 4 new log entries. - List applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 4); + List 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); @@ -382,7 +391,7 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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()); @@ -391,13 +400,15 @@ public class ReplicationAndSnapshotsIntegrationTest extends AbstractRaftActorInt 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"); }