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%2FRecoveryIntegrationTest.java;h=ca53d2e6c59658c9310dd7c559d1ea0b15226ec4;hp=33e5c4bcdf4e324ebd842c651fdbb99e8ad46f59;hb=f2868cf8e1ddc85ec1128199ed2607922ea74c24;hpb=412db94945c5db5d2da918f5e23bd3abcecc4d10 diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RecoveryIntegrationTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RecoveryIntegrationTest.java index 33e5c4bcdf..ca53d2e6c5 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RecoveryIntegrationTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RecoveryIntegrationTest.java @@ -8,15 +8,22 @@ package org.opendaylight.controller.cluster.raft; import static org.junit.Assert.assertEquals; + +import akka.actor.ActorRef; import akka.persistence.SaveSnapshotSuccess; import com.google.common.collect.ImmutableMap; import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; import org.junit.Before; import org.junit.Test; import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload; -import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries; +import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; +import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries; +import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; /** @@ -34,11 +41,12 @@ public class RecoveryIntegrationTest extends AbstractRaftActorIntegrationTest { follower1Actor = newTestRaftActor(follower1Id, ImmutableMap.of(leaderId, testActorPath(leaderId)), newFollowerConfigParams()); - peerAddresses = ImmutableMap.builder(). - put(follower1Id, follower1Actor.path().toString()).build(); + Map leaderPeerAddresses = new HashMap<>(); + leaderPeerAddresses.put(follower1Id, follower1Actor.path().toString()); + leaderPeerAddresses.put(follower2Id, ""); leaderConfigParams = newLeaderConfigParams(); - leaderActor = newTestRaftActor(leaderId, peerAddresses, leaderConfigParams); + leaderActor = newTestRaftActor(leaderId, leaderPeerAddresses, leaderConfigParams); follower1CollectorActor = follower1Actor.underlyingActor().collectorActor(); leaderCollectorActor = leaderActor.underlyingActor().collectorActor(); @@ -55,20 +63,20 @@ public class RecoveryIntegrationTest extends AbstractRaftActorIntegrationTest { leaderActor.underlyingActor().startDropMessages(CaptureSnapshotReply.class); follower1Actor.underlyingActor().startDropMessages(AppendEntries.class); - MockPayload payload2 = sendPayloadData(leaderActor, "two"); + final MockPayload payload2 = sendPayloadData(leaderActor, "two"); // This should trigger a snapshot. - MockPayload payload3 = sendPayloadData(leaderActor, "three"); + final MockPayload payload3 = sendPayloadData(leaderActor, "three"); MessageCollectorActor.expectMatching(follower1CollectorActor, AppendEntries.class, 3); // Send another payload. - MockPayload payload4 = sendPayloadData(leaderActor, "four"); + final MockPayload payload4 = sendPayloadData(leaderActor, "four"); // Now deliver the AppendEntries to the follower follower1Actor.underlyingActor().stopDropMessages(AppendEntries.class); - MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyJournalEntries.class, 3); + MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyJournalEntries.class, 1); // Now deliver the CaptureSnapshotReply to the leader. CaptureSnapshotReply captureSnapshotReply = MessageCollectorActor.expectFirstMatching( @@ -100,13 +108,13 @@ public class RecoveryIntegrationTest extends AbstractRaftActorIntegrationTest { // Block these messages initially so we can control the sequence. follower1Actor.underlyingActor().startDropMessages(AppendEntries.class); - MockPayload payload2 = sendPayloadData(leaderActor, "two"); + final MockPayload payload2 = sendPayloadData(leaderActor, "two"); // This should trigger a snapshot. - MockPayload payload3 = sendPayloadData(leaderActor, "three"); + final MockPayload payload3 = sendPayloadData(leaderActor, "three"); // Send another payload. - MockPayload payload4 = sendPayloadData(leaderActor, "four"); + final MockPayload payload4 = sendPayloadData(leaderActor, "four"); MessageCollectorActor.expectMatching(follower1CollectorActor, AppendEntries.class, 3); @@ -116,7 +124,7 @@ public class RecoveryIntegrationTest extends AbstractRaftActorIntegrationTest { // Now deliver the AppendEntries to the follower follower1Actor.underlyingActor().stopDropMessages(AppendEntries.class); - MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyJournalEntries.class, 3); + MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyJournalEntries.class, 1); reinstateLeaderActor(); @@ -131,6 +139,116 @@ public class RecoveryIntegrationTest extends AbstractRaftActorIntegrationTest { leaderActor.underlyingActor().getState()); } + @Test + public void testFollowerRecoveryAfterInstallSnapshot() { + + send2InitialPayloads(); + + leader = leaderActor.underlyingActor().getCurrentBehavior(); + + follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId)), + newFollowerConfigParams()); + follower2CollectorActor = follower2Actor.underlyingActor().collectorActor(); + + leaderActor.tell(new SetPeerAddress(follower2Id, follower2Actor.path().toString()), ActorRef.noSender()); + + final MockPayload payload2 = sendPayloadData(leaderActor, "two"); + + // Verify the leader applies the 3rd payload state. + MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyJournalEntries.class, 1); + + MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyJournalEntries.class, 1); + + assertEquals("Leader commit index", 2, leaderContext.getCommitIndex()); + assertEquals("Leader last applied", 2, leaderContext.getLastApplied()); + assertEquals("Leader snapshot index", 1, leaderContext.getReplicatedLog().getSnapshotIndex()); + assertEquals("Leader replicatedToAllIndex", 1, leader.getReplicatedToAllIndex()); + + killActor(follower2Actor); + + InMemoryJournal.clear(); + + follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId)), + newFollowerConfigParams()); + TestRaftActor follower2Underlying = follower2Actor.underlyingActor(); + follower2CollectorActor = follower2Underlying.collectorActor(); + follower2Context = follower2Underlying.getRaftActorContext(); + + leaderActor.tell(new SetPeerAddress(follower2Id, follower2Actor.path().toString()), ActorRef.noSender()); + + // The leader should install a snapshot so wait for the follower to receive ApplySnapshot. + MessageCollectorActor.expectFirstMatching(follower2CollectorActor, ApplySnapshot.class); + + // Wait for the follower to persist the snapshot. + MessageCollectorActor.expectFirstMatching(follower2CollectorActor, SaveSnapshotSuccess.class); + + final List expFollowerState = Arrays.asList(payload0, payload1, payload2); + + assertEquals("Follower commit index", 2, follower2Context.getCommitIndex()); + assertEquals("Follower last applied", 2, follower2Context.getLastApplied()); + assertEquals("Follower snapshot index", 2, follower2Context.getReplicatedLog().getSnapshotIndex()); + assertEquals("Follower state", expFollowerState, follower2Underlying.getState()); + + killActor(follower2Actor); + + follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId)), + newFollowerConfigParams()); + + follower2Underlying = follower2Actor.underlyingActor(); + follower2Underlying.waitForRecoveryComplete(); + follower2Context = follower2Underlying.getRaftActorContext(); + + assertEquals("Follower commit index", 2, follower2Context.getCommitIndex()); + assertEquals("Follower last applied", 2, follower2Context.getLastApplied()); + assertEquals("Follower snapshot index", 2, follower2Context.getReplicatedLog().getSnapshotIndex()); + assertEquals("Follower state", expFollowerState, follower2Underlying.getState()); + } + + @Test + public void testRecoveryDeleteEntries() { + send2InitialPayloads(); + + sendPayloadData(leaderActor, "two"); + + // This should trigger a snapshot. + sendPayloadData(leaderActor, "three"); + + MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class); + MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyJournalEntries.class, 2); + + // Disconnect follower from leader + killActor(follower1Actor); + + // Send another payloads + sendPayloadData(leaderActor, "four"); + sendPayloadData(leaderActor, "five"); + + verifyRaftState(leaderActor, raftState -> { + assertEquals("leader journal last index", 5, leaderContext.getReplicatedLog().lastIndex()); + }); + + // Remove entries started from 4 index + leaderActor.underlyingActor().getReplicatedLog().removeFromAndPersist(4); + + verifyRaftState(leaderActor, raftState -> { + assertEquals("leader journal last index", 3, leaderContext.getReplicatedLog().lastIndex()); + }); + + // Send new payloads + final MockPayload payload4 = sendPayloadData(leaderActor, "newFour"); + final MockPayload payload5 = sendPayloadData(leaderActor, "newFive"); + + verifyRaftState(leaderActor, raftState -> { + assertEquals("leader journal last index", 5, leaderContext.getReplicatedLog().lastIndex()); + }); + + reinstateLeaderActor(); + + assertEquals("Leader last index", 5 , leaderActor.underlyingActor().getReplicatedLog().lastIndex()); + assertEquals(payload4, leaderActor.underlyingActor().getReplicatedLog().get(4).getData()); + assertEquals(payload5, leaderActor.underlyingActor().getReplicatedLog().get(5).getData()); + } + private void reinstateLeaderActor() { killActor(leaderActor); @@ -146,6 +264,9 @@ public class RecoveryIntegrationTest extends AbstractRaftActorIntegrationTest { currentTerm = leaderContext.getTermInformation().getCurrentTerm(); payload0 = sendPayloadData(leaderActor, "zero"); + + MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyJournalEntries.class, 1); + payload1 = sendPayloadData(leaderActor, "one"); // Verify the leader applies the states.