X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2Fbehaviors%2FLeaderTest.java;h=63f94828eb4752ed6fc9367e9575f98d4bdc6d76;hb=20156b149d56b1b14a06d344366b8e1af1c22fe8;hp=895fe35bff7588526fac71e996e9120968234af2;hpb=191ad68d71f30f6ad6258ac458c60c663e5b1b85;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java index 895fe35bff..63f94828eb 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.java @@ -43,15 +43,14 @@ import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; import org.opendaylight.controller.cluster.raft.utils.DoNothingActor; import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; -import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages; import org.opendaylight.controller.protobuff.messages.cluster.raft.InstallSnapshotMessages; import scala.concurrent.duration.FiniteDuration; public class LeaderTest extends AbstractRaftActorBehaviorTest { - private ActorRef leaderActor = + private final ActorRef leaderActor = getSystem().actorOf(Props.create(DoNothingActor.class)); - private ActorRef senderActor = + private final ActorRef senderActor = getSystem().actorOf(Props.create(DoNothingActor.class)); @Test @@ -67,12 +66,12 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { }}; } - @Test public void testThatLeaderSendsAHeartbeatMessageToAllFollowers() { new JavaTestKit(getSystem()) {{ new Within(duration("1 seconds")) { + @Override protected void run() { ActorRef followerActor = getTestActor(); @@ -87,11 +86,15 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { actorContext.setPeerAddresses(peerAddresses); Leader leader = new Leader(actorContext); + leader.markFollowerActive(followerActor.path().toString()); + Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(), + TimeUnit.MILLISECONDS); leader.handleMessage(senderActor, new SendHeartBeat()); final String out = new ExpectMsg(duration("1 seconds"), "match hint") { // do not put code outside this method, will run afterwards + @Override protected String match(Object in) { Object msg = fromSerializableMessage(in); if (msg instanceof AppendEntries) { @@ -117,6 +120,7 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { new JavaTestKit(getSystem()) {{ new Within(duration("1 seconds")) { + @Override protected void run() { ActorRef followerActor = getTestActor(); @@ -132,6 +136,9 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { actorContext.setPeerAddresses(peerAddresses); Leader leader = new Leader(actorContext); + leader.markFollowerActive(followerActor.path().toString()); + Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(), + TimeUnit.MILLISECONDS); RaftActorBehavior raftBehavior = leader .handleMessage(senderActor, new Replicate(null, null, new MockRaftActorContext.MockReplicatedLogEntry(1, @@ -145,6 +152,7 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { final String out = new ExpectMsg(duration("1 seconds"), "match hint") { // do not put code outside this method, will run afterwards + @Override protected String match(Object in) { Object msg = fromSerializableMessage(in); if (msg instanceof AppendEntries) { @@ -169,6 +177,7 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { new JavaTestKit(getSystem()) {{ new Within(duration("1 seconds")) { + @Override protected void run() { ActorRef raftActor = getTestActor(); @@ -195,6 +204,7 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { new ExpectMsg(duration("1 seconds"), "match hint") { // do not put code outside this method, will run afterwards + @Override protected String match(Object in) { if (in instanceof ApplyState) { if (((ApplyState) in).getIdentifier().equals("state-id")) { @@ -266,10 +276,13 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { leader.getFollowerToSnapshot().getNextChunk(); leader.getFollowerToSnapshot().incrementChunkIndex(); + Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(), + TimeUnit.MILLISECONDS); + leader.handleMessage(leaderActor, new SendHeartBeat()); - AppendEntriesMessages.AppendEntries aeproto = (AppendEntriesMessages.AppendEntries)MessageCollectorActor.getFirstMatching( - followerActor, AppendEntries.SERIALIZABLE_CLASS); + AppendEntries aeproto = (AppendEntries)MessageCollectorActor.getFirstMatching( + followerActor, AppendEntries.class); assertNotNull("AppendEntries should be sent even if InstallSnapshotReply is not " + "received", aeproto); @@ -340,6 +353,9 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { //update follower timestamp leader.markFollowerActive(followerActor.path().toString()); + Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(), + TimeUnit.MILLISECONDS); + // this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex RaftActorBehavior raftBehavior = leader.handleMessage( senderActor, new Replicate(null, "state-id", entry)); @@ -482,6 +498,7 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { final String out = new ExpectMsg(duration("1 seconds"), "match hint") { // do not put code outside this method, will run afterwards + @Override protected String match(Object in) { if (in instanceof InstallSnapshotMessages.InstallSnapshot) { InstallSnapshot is = (InstallSnapshot) @@ -562,18 +579,18 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { assertTrue(raftBehavior instanceof Leader); - assertEquals(leader.mapFollowerToSnapshot.size(), 0); - assertEquals(leader.followerToLog.size(), 1); - assertNotNull(leader.followerToLog.get(followerActor.path().toString())); - FollowerLogInformation fli = leader.followerToLog.get(followerActor.path().toString()); - assertEquals(snapshotIndex, fli.getMatchIndex().get()); - assertEquals(snapshotIndex, fli.getMatchIndex().get()); - assertEquals(snapshotIndex + 1, fli.getNextIndex().get()); + assertEquals(0, leader.followerSnapshotSize()); + assertEquals(1, leader.followerLogSize()); + assertNotNull(leader.getFollower(followerActor.path().toString())); + FollowerLogInformation fli = leader.getFollower(followerActor.path().toString()); + assertEquals(snapshotIndex, fli.getMatchIndex()); + assertEquals(snapshotIndex, fli.getMatchIndex()); + assertEquals(snapshotIndex + 1, fli.getNextIndex()); }}; } @Test - public void testHandleInstallSnapshotReplyWithInvalidChunkIndex() throws Exception { + public void testHandleInstallSnapshotReplyWithInvalidChunkIndex() throws Exception{ new JavaTestKit(getSystem()) {{ TestActorRef followerActor = @@ -627,11 +644,15 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { assertEquals(3, installSnapshot.getTotalChunks()); - leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(), followerActor.path().toString(), -1, false)); + leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(), + followerActor.path().toString(), -1, false)); + + Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(), + TimeUnit.MILLISECONDS); leader.handleMessage(leaderActor, new SendHeartBeat()); - o = MessageCollectorActor.getAllMessages(followerActor).get(1); + o = MessageCollectorActor.getAllMatching(followerActor,InstallSnapshotMessages.InstallSnapshot.class).get(1); assertTrue(o instanceof InstallSnapshotMessages.InstallSnapshot); @@ -650,7 +671,7 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { { TestActorRef followerActor = - TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class), "follower"); + TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class), "follower-chunk"); Map peerAddresses = new HashMap<>(); peerAddresses.put(followerActor.path().toString(), @@ -704,10 +725,10 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { leader.handleMessage(followerActor, new InstallSnapshotReply(installSnapshot.getTerm(),followerActor.path().toString(),1,true )); - leader.handleMessage(leaderActor, new SendHeartBeat()); - Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS); + leader.handleMessage(leaderActor, new SendHeartBeat()); + o = MessageCollectorActor.getAllMessages(followerActor).get(1); assertTrue(o instanceof InstallSnapshotMessages.InstallSnapshot); @@ -779,6 +800,7 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { return createActorContext(leaderActor); } + @Override protected RaftActorContext createActorContext(ActorRef actorRef) { return new MockRaftActorContext("test", getSystem(), actorRef); } @@ -868,16 +890,18 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { Leader leader = new Leader(leaderActorContext); leader.markFollowerActive(followerActor.path().toString()); + Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(), + TimeUnit.MILLISECONDS); + leader.handleMessage(leaderActor, new SendHeartBeat()); - AppendEntriesMessages.AppendEntries appendEntries = - (AppendEntriesMessages.AppendEntries) MessageCollectorActor - .getFirstMatching(followerActor, AppendEntriesMessages.AppendEntries.class); + AppendEntries appendEntries = (AppendEntries) MessageCollectorActor + .getFirstMatching(followerActor, AppendEntries.class); assertNotNull(appendEntries); assertEquals(1, appendEntries.getLeaderCommit()); - assertEquals(1, appendEntries.getLogEntries(0).getIndex()); + assertEquals(1, appendEntries.getEntries().get(0).getIndex()); assertEquals(0, appendEntries.getPrevLogIndex()); AppendEntriesReply appendEntriesReply = @@ -937,16 +961,17 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { Leader leader = new Leader(leaderActorContext); leader.markFollowerActive(followerActor.path().toString()); + Thread.sleep(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis()); + leader.handleMessage(leaderActor, new SendHeartBeat()); - AppendEntriesMessages.AppendEntries appendEntries = - (AppendEntriesMessages.AppendEntries) MessageCollectorActor - .getFirstMatching(followerActor, AppendEntriesMessages.AppendEntries.class); + AppendEntries appendEntries = (AppendEntries) MessageCollectorActor + .getFirstMatching(followerActor, AppendEntries.class); assertNotNull(appendEntries); assertEquals(1, appendEntries.getLeaderCommit()); - assertEquals(1, appendEntries.getLogEntries(0).getIndex()); + assertEquals(1, appendEntries.getEntries().get(0).getIndex()); assertEquals(0, appendEntries.getPrevLogIndex()); AppendEntriesReply appendEntriesReply = @@ -1166,6 +1191,85 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { }}; } + + @Test + public void testAppendEntryCallAtEndofAppendEntryReply() throws Exception { + new JavaTestKit(getSystem()) {{ + + ActorRef leaderActor = getSystem().actorOf(Props.create(MessageCollectorActor.class)); + + MockRaftActorContext leaderActorContext = + new MockRaftActorContext("leader", getSystem(), leaderActor); + + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS)); + configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS)); + + leaderActorContext.setConfigParams(configParams); + + ActorRef followerActor = getSystem().actorOf(Props.create(ForwardMessageToBehaviorActor.class)); + + MockRaftActorContext followerActorContext = + new MockRaftActorContext("follower-reply", getSystem(), followerActor); + + followerActorContext.setConfigParams(configParams); + + Follower follower = new Follower(followerActorContext); + + ForwardMessageToBehaviorActor.setBehavior(follower); + + Map peerAddresses = new HashMap<>(); + peerAddresses.put("follower-reply", + followerActor.path().toString()); + + leaderActorContext.setPeerAddresses(peerAddresses); + + leaderActorContext.getReplicatedLog().removeFrom(0); + + //create 3 entries + leaderActorContext.setReplicatedLog( + new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build()); + + leaderActorContext.setCommitIndex(1); + + Leader leader = new Leader(leaderActorContext); + leader.markFollowerActive("follower-reply"); + + Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(), + TimeUnit.MILLISECONDS); + + leader.handleMessage(leaderActor, new SendHeartBeat()); + + AppendEntries appendEntries = (AppendEntries) ForwardMessageToBehaviorActor + .getFirstMatching(followerActor, AppendEntries.class); + + assertNotNull(appendEntries); + + assertEquals(1, appendEntries.getLeaderCommit()); + assertEquals(1, appendEntries.getEntries().get(0).getIndex()); + assertEquals(0, appendEntries.getPrevLogIndex()); + + AppendEntriesReply appendEntriesReply = + (AppendEntriesReply)ForwardMessageToBehaviorActor.getFirstMatching(leaderActor, AppendEntriesReply.class); + + assertNotNull(appendEntriesReply); + + leader.handleAppendEntriesReply(followerActor, appendEntriesReply); + + List entries = ForwardMessageToBehaviorActor + .getAllMatching(followerActor, AppendEntries.class); + + assertEquals("AppendEntries count should be 2 ", 2, entries.size()); + + AppendEntries appendEntriesSecond = (AppendEntries) entries.get(1); + + assertEquals(1, appendEntriesSecond.getLeaderCommit()); + assertEquals(2, appendEntriesSecond.getEntries().get(0).getIndex()); + assertEquals(1, appendEntriesSecond.getPrevLogIndex()); + + }}; + } + class MockLeader extends Leader { FollowerToSnapshot fts; @@ -1180,15 +1284,14 @@ public class LeaderTest extends AbstractRaftActorBehaviorTest { public void createFollowerToSnapshot(String followerId, ByteString bs ) { fts = new FollowerToSnapshot(bs); - mapFollowerToSnapshot.put(followerId, fts); - + setFollowerSnapshot(followerId, fts); } } private class MockConfigParamsImpl extends DefaultConfigParamsImpl { - private long electionTimeOutIntervalMillis; - private int snapshotChunkSize; + private final long electionTimeOutIntervalMillis; + private final int snapshotChunkSize; public MockConfigParamsImpl(long electionTimeOutIntervalMillis, int snapshotChunkSize) { super();