95ec0a6f2fb5fa126be7ae7f1ca819911c97cad2
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / behaviors / LeaderTest.java
1 package org.opendaylight.controller.cluster.raft.behaviors;
2
3 import static org.junit.Assert.assertEquals;
4 import static org.junit.Assert.assertNotNull;
5 import static org.junit.Assert.assertTrue;
6 import akka.actor.ActorRef;
7 import akka.actor.PoisonPill;
8 import akka.actor.Props;
9 import akka.actor.Terminated;
10 import akka.testkit.JavaTestKit;
11 import akka.testkit.TestActorRef;
12 import com.google.common.base.Optional;
13 import com.google.common.util.concurrent.Uninterruptibles;
14 import com.google.protobuf.ByteString;
15 import java.io.ByteArrayOutputStream;
16 import java.io.IOException;
17 import java.io.ObjectOutputStream;
18 import java.util.HashMap;
19 import java.util.List;
20 import java.util.Map;
21 import java.util.concurrent.TimeUnit;
22 import org.junit.Assert;
23 import org.junit.Test;
24 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
25 import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
26 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
27 import org.opendaylight.controller.cluster.raft.RaftActorContext;
28 import org.opendaylight.controller.cluster.raft.RaftState;
29 import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
30 import org.opendaylight.controller.cluster.raft.SerializationUtils;
31 import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
32 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
33 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
34 import org.opendaylight.controller.cluster.raft.base.messages.InitiateInstallSnapshot;
35 import org.opendaylight.controller.cluster.raft.base.messages.IsolatedLeaderCheck;
36 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
37 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
38 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
39 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
40 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
41 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
42 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
43 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
44 import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
45 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
46 import org.opendaylight.controller.protobuff.messages.cluster.raft.InstallSnapshotMessages;
47 import scala.concurrent.duration.FiniteDuration;
48
49 public class LeaderTest extends AbstractRaftActorBehaviorTest {
50
51     private final ActorRef leaderActor =
52         getSystem().actorOf(Props.create(DoNothingActor.class));
53     private final ActorRef senderActor =
54         getSystem().actorOf(Props.create(DoNothingActor.class));
55
56     @Test
57     public void testHandleMessageForUnknownMessage() throws Exception {
58         new JavaTestKit(getSystem()) {{
59             Leader leader =
60                 new Leader(createActorContext());
61
62             // handle message should return the Leader state when it receives an
63             // unknown message
64             RaftActorBehavior behavior = leader.handleMessage(senderActor, "foo");
65             Assert.assertTrue(behavior instanceof Leader);
66         }};
67     }
68
69     @Test
70     public void testThatLeaderSendsAHeartbeatMessageToAllFollowers() {
71         new JavaTestKit(getSystem()) {{
72
73             new Within(duration("1 seconds")) {
74                 @Override
75                 protected void run() {
76
77                     ActorRef followerActor = getTestActor();
78
79                     MockRaftActorContext actorContext = (MockRaftActorContext) createActorContext();
80
81                     Map<String, String> peerAddresses = new HashMap<>();
82
83                     peerAddresses.put(followerActor.path().toString(),
84                         followerActor.path().toString());
85
86                     actorContext.setPeerAddresses(peerAddresses);
87
88                     Leader leader = new Leader(actorContext);
89                     leader.markFollowerActive(followerActor.path().toString());
90                     Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
91                         TimeUnit.MILLISECONDS);
92                     leader.handleMessage(senderActor, new SendHeartBeat());
93
94                     final String out =
95                         new ExpectMsg<String>(duration("1 seconds"), "match hint") {
96                             // do not put code outside this method, will run afterwards
97                             @Override
98                             protected String match(Object in) {
99                                 Object msg = fromSerializableMessage(in);
100                                 if (msg instanceof AppendEntries) {
101                                     if (((AppendEntries)msg).getTerm() == 0) {
102                                         return "match";
103                                     }
104                                     return null;
105                                 } else {
106                                     throw noMatch();
107                                 }
108                             }
109                         }.get(); // this extracts the received message
110
111                     assertEquals("match", out);
112
113                 }
114             };
115         }};
116     }
117
118     @Test
119     public void testHandleReplicateMessageSendAppendEntriesToFollower() {
120         new JavaTestKit(getSystem()) {{
121
122             new Within(duration("1 seconds")) {
123                 @Override
124                 protected void run() {
125
126                     ActorRef followerActor = getTestActor();
127
128                     MockRaftActorContext actorContext =
129                         (MockRaftActorContext) createActorContext();
130
131                     Map<String, String> peerAddresses = new HashMap<>();
132
133                     peerAddresses.put(followerActor.path().toString(),
134                             followerActor.path().toString());
135
136                     actorContext.setPeerAddresses(peerAddresses);
137
138                     Leader leader = new Leader(actorContext);
139                     leader.markFollowerActive(followerActor.path().toString());
140                     Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
141                         TimeUnit.MILLISECONDS);
142                     RaftActorBehavior raftBehavior = leader
143                         .handleMessage(senderActor, new Replicate(null, null,
144                             new MockRaftActorContext.MockReplicatedLogEntry(1,
145                                 100,
146                                 new MockRaftActorContext.MockPayload("foo"))
147                         ));
148
149                     // State should not change
150                     assertTrue(raftBehavior instanceof Leader);
151
152                     final String out =
153                         new ExpectMsg<String>(duration("1 seconds"), "match hint") {
154                             // do not put code outside this method, will run afterwards
155                             @Override
156                             protected String match(Object in) {
157                                 Object msg = fromSerializableMessage(in);
158                                 if (msg instanceof AppendEntries) {
159                                     if (((AppendEntries)msg).getTerm() == 0) {
160                                         return "match";
161                                     }
162                                     return null;
163                                 } else {
164                                     throw noMatch();
165                                 }
166                             }
167                         }.get(); // this extracts the received message
168
169                     assertEquals("match", out);
170                 }
171             };
172         }};
173     }
174
175     @Test
176     public void testHandleReplicateMessageWhenThereAreNoFollowers() {
177         new JavaTestKit(getSystem()) {{
178
179             new Within(duration("1 seconds")) {
180                 @Override
181                 protected void run() {
182
183                     ActorRef raftActor = getTestActor();
184
185                     MockRaftActorContext actorContext =
186                         new MockRaftActorContext("test", getSystem(), raftActor);
187
188                     actorContext.getReplicatedLog().removeFrom(0);
189
190                     actorContext.setReplicatedLog(
191                         new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1)
192                             .build());
193
194                     Leader leader = new Leader(actorContext);
195                     RaftActorBehavior raftBehavior = leader
196                         .handleMessage(senderActor, new Replicate(null, "state-id",actorContext.getReplicatedLog().get(1)));
197
198                     // State should not change
199                     assertTrue(raftBehavior instanceof Leader);
200
201                     assertEquals(1, actorContext.getCommitIndex());
202
203                     final String out =
204                         new ExpectMsg<String>(duration("1 seconds"),
205                             "match hint") {
206                             // do not put code outside this method, will run afterwards
207                             @Override
208                             protected String match(Object in) {
209                                 if (in instanceof ApplyState) {
210                                     if (((ApplyState) in).getIdentifier().equals("state-id")) {
211                                         return "match";
212                                     }
213                                     return null;
214                                 } else {
215                                     throw noMatch();
216                                 }
217                             }
218                         }.get(); // this extracts the received message
219
220                     assertEquals("match", out);
221
222                 }
223             };
224         }};
225     }
226
227     @Test
228     public void testSendAppendEntriesOnAnInProgressInstallSnapshot() throws Exception {
229         new JavaTestKit(getSystem()) {{
230             ActorRef followerActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
231
232             Map<String, String> peerAddresses = new HashMap<>();
233             peerAddresses.put(followerActor.path().toString(),
234                 followerActor.path().toString());
235
236             MockRaftActorContext actorContext =
237                 (MockRaftActorContext) createActorContext(leaderActor);
238             actorContext.setPeerAddresses(peerAddresses);
239
240             Map<String, String> leadersSnapshot = new HashMap<>();
241             leadersSnapshot.put("1", "A");
242             leadersSnapshot.put("2", "B");
243             leadersSnapshot.put("3", "C");
244
245             //clears leaders log
246             actorContext.getReplicatedLog().removeFrom(0);
247
248             final int followersLastIndex = 2;
249             final int snapshotIndex = 3;
250             final int newEntryIndex = 4;
251             final int snapshotTerm = 1;
252             final int currentTerm = 2;
253
254             // set the snapshot variables in replicatedlog
255             actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
256             actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
257             actorContext.setCommitIndex(followersLastIndex);
258             //set follower timeout to 2 mins, helps during debugging
259             actorContext.setConfigParams(new MockConfigParamsImpl(120000L, 10));
260
261             MockLeader leader = new MockLeader(actorContext);
262
263             // new entry
264             ReplicatedLogImplEntry entry =
265                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
266                     new MockRaftActorContext.MockPayload("D"));
267
268             //update follower timestamp
269             leader.markFollowerActive(followerActor.path().toString());
270
271             ByteString bs = toByteString(leadersSnapshot);
272             leader.setSnapshot(Optional.of(bs));
273             leader.createFollowerToSnapshot(followerActor.path().toString(), bs);
274
275             //send first chunk and no InstallSnapshotReply received yet
276             leader.getFollowerToSnapshot().getNextChunk();
277             leader.getFollowerToSnapshot().incrementChunkIndex();
278
279             Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
280                 TimeUnit.MILLISECONDS);
281
282             leader.handleMessage(leaderActor, new SendHeartBeat());
283
284             AppendEntries aeproto = (AppendEntries)MessageCollectorActor.getFirstMatching(
285                 followerActor, AppendEntries.class);
286
287             assertNotNull("AppendEntries should be sent even if InstallSnapshotReply is not " +
288                 "received", aeproto);
289
290             AppendEntries ae = (AppendEntries) SerializationUtils.fromSerializable(aeproto);
291
292             assertTrue("AppendEntries should be sent with empty entries", ae.getEntries().isEmpty());
293
294             //InstallSnapshotReply received
295             leader.getFollowerToSnapshot().markSendStatus(true);
296
297             leader.handleMessage(senderActor, new SendHeartBeat());
298
299             InstallSnapshotMessages.InstallSnapshot isproto = (InstallSnapshotMessages.InstallSnapshot)
300                 MessageCollectorActor.getFirstMatching(followerActor,
301                     InstallSnapshot.SERIALIZABLE_CLASS);
302
303             assertNotNull("Installsnapshot should get called for sending the next chunk of snapshot",
304                 isproto);
305
306             InstallSnapshot is = (InstallSnapshot) SerializationUtils.fromSerializable(isproto);
307
308             assertEquals(snapshotIndex, is.getLastIncludedIndex());
309
310         }};
311     }
312
313     @Test
314     public void testSendAppendEntriesSnapshotScenario() {
315         new JavaTestKit(getSystem()) {{
316
317             ActorRef followerActor = getTestActor();
318
319             Map<String, String> peerAddresses = new HashMap<>();
320             peerAddresses.put(followerActor.path().toString(),
321                 followerActor.path().toString());
322
323             MockRaftActorContext actorContext =
324                 (MockRaftActorContext) createActorContext(getRef());
325             actorContext.setPeerAddresses(peerAddresses);
326
327             Map<String, String> leadersSnapshot = new HashMap<>();
328             leadersSnapshot.put("1", "A");
329             leadersSnapshot.put("2", "B");
330             leadersSnapshot.put("3", "C");
331
332             //clears leaders log
333             actorContext.getReplicatedLog().removeFrom(0);
334
335             final int followersLastIndex = 2;
336             final int snapshotIndex = 3;
337             final int newEntryIndex = 4;
338             final int snapshotTerm = 1;
339             final int currentTerm = 2;
340
341             // set the snapshot variables in replicatedlog
342             actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
343             actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
344             actorContext.setCommitIndex(followersLastIndex);
345
346             Leader leader = new Leader(actorContext);
347
348             // new entry
349             ReplicatedLogImplEntry entry =
350                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
351                     new MockRaftActorContext.MockPayload("D"));
352
353             //update follower timestamp
354             leader.markFollowerActive(followerActor.path().toString());
355
356             Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
357                 TimeUnit.MILLISECONDS);
358
359             // this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex
360             RaftActorBehavior raftBehavior = leader.handleMessage(
361                 senderActor, new Replicate(null, "state-id", entry));
362
363             assertTrue(raftBehavior instanceof Leader);
364
365             // we might receive some heartbeat messages, so wait till we InitiateInstallSnapshot
366             Boolean[] matches = new ReceiveWhile<Boolean>(Boolean.class, duration("2 seconds")) {
367                 @Override
368                 protected Boolean match(Object o) throws Exception {
369                     if (o instanceof InitiateInstallSnapshot) {
370                         return true;
371                     }
372                     return false;
373                 }
374             }.get();
375
376             boolean initiateInitiateInstallSnapshot = false;
377             for (Boolean b: matches) {
378                 initiateInitiateInstallSnapshot = b | initiateInitiateInstallSnapshot;
379             }
380
381             assertTrue(initiateInitiateInstallSnapshot);
382         }};
383     }
384
385     @Test
386     public void testInitiateInstallSnapshot() throws Exception {
387         new JavaTestKit(getSystem()) {{
388
389             ActorRef leaderActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
390
391             ActorRef followerActor = getTestActor();
392
393             Map<String, String> peerAddresses = new HashMap<>();
394             peerAddresses.put(followerActor.path().toString(),
395                 followerActor.path().toString());
396
397
398             MockRaftActorContext actorContext =
399                 (MockRaftActorContext) createActorContext(leaderActor);
400             actorContext.setPeerAddresses(peerAddresses);
401
402             Map<String, String> leadersSnapshot = new HashMap<>();
403             leadersSnapshot.put("1", "A");
404             leadersSnapshot.put("2", "B");
405             leadersSnapshot.put("3", "C");
406
407             //clears leaders log
408             actorContext.getReplicatedLog().removeFrom(0);
409
410             final int followersLastIndex = 2;
411             final int snapshotIndex = 3;
412             final int newEntryIndex = 4;
413             final int snapshotTerm = 1;
414             final int currentTerm = 2;
415
416             // set the snapshot variables in replicatedlog
417             actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
418             actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
419             actorContext.setLastApplied(3);
420             actorContext.setCommitIndex(followersLastIndex);
421
422             Leader leader = new Leader(actorContext);
423             // set the snapshot as absent and check if capture-snapshot is invoked.
424             leader.setSnapshot(Optional.<ByteString>absent());
425
426             // new entry
427             ReplicatedLogImplEntry entry =
428                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
429                     new MockRaftActorContext.MockPayload("D"));
430
431             actorContext.getReplicatedLog().append(entry);
432
433             // this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex
434             RaftActorBehavior raftBehavior = leader.handleMessage(
435                 leaderActor, new InitiateInstallSnapshot());
436
437             CaptureSnapshot cs = (CaptureSnapshot) MessageCollectorActor.
438                 getFirstMatching(leaderActor, CaptureSnapshot.class);
439
440             assertNotNull(cs);
441
442             assertTrue(cs.isInstallSnapshotInitiated());
443             assertEquals(3, cs.getLastAppliedIndex());
444             assertEquals(1, cs.getLastAppliedTerm());
445             assertEquals(4, cs.getLastIndex());
446             assertEquals(2, cs.getLastTerm());
447         }};
448     }
449
450     @Test
451     public void testInstallSnapshot() {
452         new JavaTestKit(getSystem()) {{
453
454             ActorRef followerActor = getTestActor();
455
456             Map<String, String> peerAddresses = new HashMap<>();
457             peerAddresses.put(followerActor.path().toString(),
458                 followerActor.path().toString());
459
460             MockRaftActorContext actorContext =
461                 (MockRaftActorContext) createActorContext();
462             actorContext.setPeerAddresses(peerAddresses);
463
464
465             Map<String, String> leadersSnapshot = new HashMap<>();
466             leadersSnapshot.put("1", "A");
467             leadersSnapshot.put("2", "B");
468             leadersSnapshot.put("3", "C");
469
470             //clears leaders log
471             actorContext.getReplicatedLog().removeFrom(0);
472
473             final int followersLastIndex = 2;
474             final int snapshotIndex = 3;
475             final int newEntryIndex = 4;
476             final int snapshotTerm = 1;
477             final int currentTerm = 2;
478
479             // set the snapshot variables in replicatedlog
480             actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
481             actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
482             actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
483             actorContext.setCommitIndex(followersLastIndex);
484
485             Leader leader = new Leader(actorContext);
486
487             // new entry
488             ReplicatedLogImplEntry entry =
489                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
490                     new MockRaftActorContext.MockPayload("D"));
491
492             RaftActorBehavior raftBehavior = leader.handleMessage(senderActor,
493                 new SendInstallSnapshot(toByteString(leadersSnapshot)));
494
495             assertTrue(raftBehavior instanceof Leader);
496
497             // check if installsnapshot gets called with the correct values.
498             final String out =
499                 new ExpectMsg<String>(duration("1 seconds"), "match hint") {
500                     // do not put code outside this method, will run afterwards
501                     @Override
502                     protected String match(Object in) {
503                         if (in instanceof InstallSnapshotMessages.InstallSnapshot) {
504                             InstallSnapshot is = (InstallSnapshot)
505                                 SerializationUtils.fromSerializable(in);
506                             if (is.getData() == null) {
507                                 return "InstallSnapshot data is null";
508                             }
509                             if (is.getLastIncludedIndex() != snapshotIndex) {
510                                 return is.getLastIncludedIndex() + "!=" + snapshotIndex;
511                             }
512                             if (is.getLastIncludedTerm() != snapshotTerm) {
513                                 return is.getLastIncludedTerm() + "!=" + snapshotTerm;
514                             }
515                             if (is.getTerm() == currentTerm) {
516                                 return is.getTerm() + "!=" + currentTerm;
517                             }
518
519                             return "match";
520
521                         } else {
522                             return "message mismatch:" + in.getClass();
523                         }
524                     }
525                 }.get(); // this extracts the received message
526
527             assertEquals("match", out);
528         }};
529     }
530
531     @Test
532     public void testHandleInstallSnapshotReplyLastChunk() {
533         new JavaTestKit(getSystem()) {{
534
535             ActorRef followerActor = getTestActor();
536
537             Map<String, String> peerAddresses = new HashMap<>();
538             peerAddresses.put(followerActor.path().toString(),
539                 followerActor.path().toString());
540
541             final int followersLastIndex = 2;
542             final int snapshotIndex = 3;
543             final int newEntryIndex = 4;
544             final int snapshotTerm = 1;
545             final int currentTerm = 2;
546
547             MockRaftActorContext actorContext =
548                 (MockRaftActorContext) createActorContext();
549             actorContext.setPeerAddresses(peerAddresses);
550             actorContext.setCommitIndex(followersLastIndex);
551
552             MockLeader leader = new MockLeader(actorContext);
553
554             Map<String, String> leadersSnapshot = new HashMap<>();
555             leadersSnapshot.put("1", "A");
556             leadersSnapshot.put("2", "B");
557             leadersSnapshot.put("3", "C");
558
559             // set the snapshot variables in replicatedlog
560
561             actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
562             actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
563             actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
564
565             ByteString bs = toByteString(leadersSnapshot);
566             leader.setSnapshot(Optional.of(bs));
567             leader.createFollowerToSnapshot(followerActor.path().toString(), bs);
568             while(!leader.getFollowerToSnapshot().isLastChunk(leader.getFollowerToSnapshot().getChunkIndex())) {
569                 leader.getFollowerToSnapshot().getNextChunk();
570                 leader.getFollowerToSnapshot().incrementChunkIndex();
571             }
572
573             //clears leaders log
574             actorContext.getReplicatedLog().removeFrom(0);
575
576             RaftActorBehavior raftBehavior = leader.handleMessage(senderActor,
577                 new InstallSnapshotReply(currentTerm, followerActor.path().toString(),
578                     leader.getFollowerToSnapshot().getChunkIndex(), true));
579
580             assertTrue(raftBehavior instanceof Leader);
581
582             assertEquals(0, leader.followerSnapshotSize());
583             assertEquals(1, leader.followerLogSize());
584             assertNotNull(leader.getFollower(followerActor.path().toString()));
585             FollowerLogInformation fli = leader.getFollower(followerActor.path().toString());
586             assertEquals(snapshotIndex, fli.getMatchIndex());
587             assertEquals(snapshotIndex, fli.getMatchIndex());
588             assertEquals(snapshotIndex + 1, fli.getNextIndex());
589         }};
590     }
591     @Test
592     public void testSendSnapshotfromInstallSnapshotReply() throws Exception {
593         new JavaTestKit(getSystem()) {{
594
595             TestActorRef<MessageCollectorActor> followerActor =
596                 TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class), "follower-reply");
597
598             Map<String, String> peerAddresses = new HashMap<>();
599             peerAddresses.put("follower-reply",
600                 followerActor.path().toString());
601
602             final int followersLastIndex = 2;
603             final int snapshotIndex = 3;
604             final int snapshotTerm = 1;
605             final int currentTerm = 2;
606
607             MockRaftActorContext actorContext =
608                 (MockRaftActorContext) createActorContext();
609             DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(){
610                 @Override
611                 public int getSnapshotChunkSize() {
612                     return 50;
613                 }
614             };
615             configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS));
616             configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
617
618             actorContext.setConfigParams(configParams);
619             actorContext.setPeerAddresses(peerAddresses);
620             actorContext.setCommitIndex(followersLastIndex);
621
622             MockLeader leader = new MockLeader(actorContext);
623
624             Map<String, String> leadersSnapshot = new HashMap<>();
625             leadersSnapshot.put("1", "A");
626             leadersSnapshot.put("2", "B");
627             leadersSnapshot.put("3", "C");
628
629             // set the snapshot variables in replicatedlog
630             actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
631             actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
632             actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
633
634             ByteString bs = toByteString(leadersSnapshot);
635             leader.setSnapshot(Optional.of(bs));
636
637             leader.handleMessage(leaderActor, new SendInstallSnapshot(bs));
638
639             List<Object> objectList = MessageCollectorActor.getAllMatching(followerActor,
640                 InstallSnapshotMessages.InstallSnapshot.class);
641
642             assertEquals(1, objectList.size());
643
644             Object o = objectList.get(0);
645             assertTrue(o instanceof InstallSnapshotMessages.InstallSnapshot);
646
647             InstallSnapshotMessages.InstallSnapshot installSnapshot = (InstallSnapshotMessages.InstallSnapshot) o;
648
649             assertEquals(1, installSnapshot.getChunkIndex());
650             assertEquals(3, installSnapshot.getTotalChunks());
651
652             leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
653                 "follower-reply", installSnapshot.getChunkIndex(), true));
654
655             objectList = MessageCollectorActor.getAllMatching(followerActor,
656                 InstallSnapshotMessages.InstallSnapshot.class);
657
658             assertEquals(2, objectList.size());
659
660             installSnapshot = (InstallSnapshotMessages.InstallSnapshot) objectList.get(1);
661
662             leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
663                 "follower-reply", installSnapshot.getChunkIndex(), true));
664
665             objectList = MessageCollectorActor.getAllMatching(followerActor,
666                 InstallSnapshotMessages.InstallSnapshot.class);
667
668             assertEquals(3, objectList.size());
669
670             installSnapshot = (InstallSnapshotMessages.InstallSnapshot) objectList.get(2);
671
672             // Send snapshot reply one more time and make sure that a new snapshot message should not be sent to follower
673             leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
674                 "follower-reply", installSnapshot.getChunkIndex(), true));
675
676             objectList = MessageCollectorActor.getAllMatching(followerActor,
677                 InstallSnapshotMessages.InstallSnapshot.class);
678
679             // Count should still stay at 3
680             assertEquals(3, objectList.size());
681         }};
682     }
683
684
685     @Test
686     public void testHandleInstallSnapshotReplyWithInvalidChunkIndex() throws Exception{
687         new JavaTestKit(getSystem()) {{
688
689             TestActorRef<MessageCollectorActor> followerActor =
690                     TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class), "follower");
691
692             Map<String, String> peerAddresses = new HashMap<>();
693             peerAddresses.put(followerActor.path().toString(),
694                     followerActor.path().toString());
695
696             final int followersLastIndex = 2;
697             final int snapshotIndex = 3;
698             final int snapshotTerm = 1;
699             final int currentTerm = 2;
700
701             MockRaftActorContext actorContext =
702                     (MockRaftActorContext) createActorContext();
703
704             actorContext.setConfigParams(new DefaultConfigParamsImpl(){
705                 @Override
706                 public int getSnapshotChunkSize() {
707                     return 50;
708                 }
709             });
710             actorContext.setPeerAddresses(peerAddresses);
711             actorContext.setCommitIndex(followersLastIndex);
712
713             MockLeader leader = new MockLeader(actorContext);
714
715             Map<String, String> leadersSnapshot = new HashMap<>();
716             leadersSnapshot.put("1", "A");
717             leadersSnapshot.put("2", "B");
718             leadersSnapshot.put("3", "C");
719
720             // set the snapshot variables in replicatedlog
721             actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
722             actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
723             actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
724
725             ByteString bs = toByteString(leadersSnapshot);
726             leader.setSnapshot(Optional.of(bs));
727
728             leader.handleMessage(leaderActor, new SendInstallSnapshot(bs));
729
730             Object o = MessageCollectorActor.getAllMessages(followerActor).get(0);
731
732             assertTrue(o instanceof InstallSnapshotMessages.InstallSnapshot);
733
734             InstallSnapshotMessages.InstallSnapshot installSnapshot = (InstallSnapshotMessages.InstallSnapshot) o;
735
736             assertEquals(1, installSnapshot.getChunkIndex());
737             assertEquals(3, installSnapshot.getTotalChunks());
738
739
740             leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
741                 followerActor.path().toString(), -1, false));
742
743             Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
744                 TimeUnit.MILLISECONDS);
745
746             leader.handleMessage(leaderActor, new SendHeartBeat());
747
748             o = MessageCollectorActor.getAllMatching(followerActor,InstallSnapshotMessages.InstallSnapshot.class).get(1);
749
750             assertTrue(o instanceof InstallSnapshotMessages.InstallSnapshot);
751
752             installSnapshot = (InstallSnapshotMessages.InstallSnapshot) o;
753
754             assertEquals(1, installSnapshot.getChunkIndex());
755             assertEquals(3, installSnapshot.getTotalChunks());
756
757             followerActor.tell(PoisonPill.getInstance(), getRef());
758         }};
759     }
760
761     @Test
762     public void testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk() throws Exception {
763         new JavaTestKit(getSystem()) {
764             {
765
766                 TestActorRef<MessageCollectorActor> followerActor =
767                         TestActorRef.create(getSystem(), Props.create(MessageCollectorActor.class), "follower-chunk");
768
769                 Map<String, String> peerAddresses = new HashMap<>();
770                 peerAddresses.put(followerActor.path().toString(),
771                         followerActor.path().toString());
772
773                 final int followersLastIndex = 2;
774                 final int snapshotIndex = 3;
775                 final int snapshotTerm = 1;
776                 final int currentTerm = 2;
777
778                 MockRaftActorContext actorContext =
779                         (MockRaftActorContext) createActorContext();
780
781                 actorContext.setConfigParams(new DefaultConfigParamsImpl() {
782                     @Override
783                     public int getSnapshotChunkSize() {
784                         return 50;
785                     }
786                 });
787                 actorContext.setPeerAddresses(peerAddresses);
788                 actorContext.setCommitIndex(followersLastIndex);
789
790                 MockLeader leader = new MockLeader(actorContext);
791
792                 Map<String, String> leadersSnapshot = new HashMap<>();
793                 leadersSnapshot.put("1", "A");
794                 leadersSnapshot.put("2", "B");
795                 leadersSnapshot.put("3", "C");
796
797                 // set the snapshot variables in replicatedlog
798                 actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
799                 actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
800                 actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
801
802                 ByteString bs = toByteString(leadersSnapshot);
803                 leader.setSnapshot(Optional.of(bs));
804
805                 leader.handleMessage(leaderActor, new SendInstallSnapshot(bs));
806
807                 Object o = MessageCollectorActor.getAllMessages(followerActor).get(0);
808
809                 assertTrue(o instanceof InstallSnapshotMessages.InstallSnapshot);
810
811                 InstallSnapshotMessages.InstallSnapshot installSnapshot = (InstallSnapshotMessages.InstallSnapshot) o;
812
813                 assertEquals(1, installSnapshot.getChunkIndex());
814                 assertEquals(3, installSnapshot.getTotalChunks());
815                 assertEquals(AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE, installSnapshot.getLastChunkHashCode());
816
817                 int hashCode = installSnapshot.getData().hashCode();
818
819                 leader.handleMessage(followerActor, new InstallSnapshotReply(installSnapshot.getTerm(),followerActor.path().toString(),1,true ));
820
821                 Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
822
823                 leader.handleMessage(leaderActor, new SendHeartBeat());
824
825                 o = MessageCollectorActor.getAllMessages(followerActor).get(1);
826
827                 assertTrue(o instanceof InstallSnapshotMessages.InstallSnapshot);
828
829                 installSnapshot = (InstallSnapshotMessages.InstallSnapshot) o;
830
831                 assertEquals(2, installSnapshot.getChunkIndex());
832                 assertEquals(3, installSnapshot.getTotalChunks());
833                 assertEquals(hashCode, installSnapshot.getLastChunkHashCode());
834
835                 followerActor.tell(PoisonPill.getInstance(), getRef());
836             }};
837     }
838
839     @Test
840     public void testFollowerToSnapshotLogic() {
841
842         MockRaftActorContext actorContext = (MockRaftActorContext) createActorContext();
843
844         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
845             @Override
846             public int getSnapshotChunkSize() {
847                 return 50;
848             }
849         });
850
851         MockLeader leader = new MockLeader(actorContext);
852
853         Map<String, String> leadersSnapshot = new HashMap<>();
854         leadersSnapshot.put("1", "A");
855         leadersSnapshot.put("2", "B");
856         leadersSnapshot.put("3", "C");
857
858         ByteString bs = toByteString(leadersSnapshot);
859         byte[] barray = bs.toByteArray();
860
861         leader.createFollowerToSnapshot("followerId", bs);
862         assertEquals(bs.size(), barray.length);
863
864         int chunkIndex=0;
865         for (int i=0; i < barray.length; i = i + 50) {
866             int j = i + 50;
867             chunkIndex++;
868
869             if (i + 50 > barray.length) {
870                 j = barray.length;
871             }
872
873             ByteString chunk = leader.getFollowerToSnapshot().getNextChunk();
874             assertEquals("bytestring size not matching for chunk:"+ chunkIndex, j-i, chunk.size());
875             assertEquals("chunkindex not matching", chunkIndex, leader.getFollowerToSnapshot().getChunkIndex());
876
877             leader.getFollowerToSnapshot().markSendStatus(true);
878             if (!leader.getFollowerToSnapshot().isLastChunk(chunkIndex)) {
879                 leader.getFollowerToSnapshot().incrementChunkIndex();
880             }
881         }
882
883         assertEquals("totalChunks not matching", chunkIndex, leader.getFollowerToSnapshot().getTotalChunks());
884     }
885
886
887     @Override protected RaftActorBehavior createBehavior(
888         RaftActorContext actorContext) {
889         return new Leader(actorContext);
890     }
891
892     @Override protected RaftActorContext createActorContext() {
893         return createActorContext(leaderActor);
894     }
895
896     @Override
897     protected RaftActorContext createActorContext(ActorRef actorRef) {
898         return new MockRaftActorContext("test", getSystem(), actorRef);
899     }
900
901     private ByteString toByteString(Map<String, String> state) {
902         ByteArrayOutputStream b = null;
903         ObjectOutputStream o = null;
904         try {
905             try {
906                 b = new ByteArrayOutputStream();
907                 o = new ObjectOutputStream(b);
908                 o.writeObject(state);
909                 byte[] snapshotBytes = b.toByteArray();
910                 return ByteString.copyFrom(snapshotBytes);
911             } finally {
912                 if (o != null) {
913                     o.flush();
914                     o.close();
915                 }
916                 if (b != null) {
917                     b.close();
918                 }
919             }
920         } catch (IOException e) {
921             Assert.fail("IOException in converting Hashmap to Bytestring:" + e);
922         }
923         return null;
924     }
925
926     public static class ForwardMessageToBehaviorActor extends MessageCollectorActor {
927         private static AbstractRaftActorBehavior behavior;
928
929         public ForwardMessageToBehaviorActor(){
930
931         }
932
933         @Override public void onReceive(Object message) throws Exception {
934             super.onReceive(message);
935             behavior.handleMessage(sender(), message);
936         }
937
938         public static void setBehavior(AbstractRaftActorBehavior behavior){
939             ForwardMessageToBehaviorActor.behavior = behavior;
940         }
941     }
942
943     @Test
944     public void testLeaderCreatedWithCommitIndexLessThanLastIndex() throws Exception {
945         new JavaTestKit(getSystem()) {{
946
947             ActorRef leaderActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
948
949             MockRaftActorContext leaderActorContext =
950                 new MockRaftActorContext("leader", getSystem(), leaderActor);
951
952             ActorRef followerActor = getSystem().actorOf(Props.create(ForwardMessageToBehaviorActor.class));
953
954             MockRaftActorContext followerActorContext =
955                 new MockRaftActorContext("follower", getSystem(), followerActor);
956
957             Follower follower = new Follower(followerActorContext);
958
959             ForwardMessageToBehaviorActor.setBehavior(follower);
960
961             Map<String, String> peerAddresses = new HashMap<>();
962             peerAddresses.put(followerActor.path().toString(),
963                 followerActor.path().toString());
964
965             leaderActorContext.setPeerAddresses(peerAddresses);
966
967             leaderActorContext.getReplicatedLog().removeFrom(0);
968
969             //create 3 entries
970             leaderActorContext.setReplicatedLog(
971                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
972
973             leaderActorContext.setCommitIndex(1);
974
975             followerActorContext.getReplicatedLog().removeFrom(0);
976
977             // follower too has the exact same log entries and has the same commit index
978             followerActorContext.setReplicatedLog(
979                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
980
981             followerActorContext.setCommitIndex(1);
982
983             Leader leader = new Leader(leaderActorContext);
984             leader.markFollowerActive(followerActor.path().toString());
985
986             Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(),
987                 TimeUnit.MILLISECONDS);
988
989             leader.handleMessage(leaderActor, new SendHeartBeat());
990
991             AppendEntries appendEntries = (AppendEntries) MessageCollectorActor
992                     .getFirstMatching(followerActor, AppendEntries.class);
993
994             assertNotNull(appendEntries);
995
996             assertEquals(1, appendEntries.getLeaderCommit());
997             assertEquals(1, appendEntries.getEntries().get(0).getIndex());
998             assertEquals(0, appendEntries.getPrevLogIndex());
999
1000             AppendEntriesReply appendEntriesReply =
1001                 (AppendEntriesReply) MessageCollectorActor.getFirstMatching(
1002                     leaderActor, AppendEntriesReply.class);
1003
1004             assertNotNull(appendEntriesReply);
1005
1006             // follower returns its next index
1007             assertEquals(2, appendEntriesReply.getLogLastIndex());
1008             assertEquals(1, appendEntriesReply.getLogLastTerm());
1009
1010         }};
1011     }
1012
1013
1014     @Test
1015     public void testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex() throws Exception {
1016         new JavaTestKit(getSystem()) {{
1017
1018             ActorRef leaderActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
1019
1020             MockRaftActorContext leaderActorContext =
1021                 new MockRaftActorContext("leader", getSystem(), leaderActor);
1022
1023             ActorRef followerActor = getSystem().actorOf(
1024                 Props.create(ForwardMessageToBehaviorActor.class));
1025
1026             MockRaftActorContext followerActorContext =
1027                 new MockRaftActorContext("follower", getSystem(), followerActor);
1028
1029             Follower follower = new Follower(followerActorContext);
1030
1031             ForwardMessageToBehaviorActor.setBehavior(follower);
1032
1033             Map<String, String> peerAddresses = new HashMap<>();
1034             peerAddresses.put(followerActor.path().toString(),
1035                 followerActor.path().toString());
1036
1037             leaderActorContext.setPeerAddresses(peerAddresses);
1038
1039             leaderActorContext.getReplicatedLog().removeFrom(0);
1040
1041             leaderActorContext.setReplicatedLog(
1042                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1043
1044             leaderActorContext.setCommitIndex(1);
1045
1046             followerActorContext.getReplicatedLog().removeFrom(0);
1047
1048             followerActorContext.setReplicatedLog(
1049                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1050
1051             // follower has the same log entries but its commit index > leaders commit index
1052             followerActorContext.setCommitIndex(2);
1053
1054             Leader leader = new Leader(leaderActorContext);
1055             leader.markFollowerActive(followerActor.path().toString());
1056
1057             Thread.sleep(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis());
1058
1059             leader.handleMessage(leaderActor, new SendHeartBeat());
1060
1061             AppendEntries appendEntries = (AppendEntries) MessageCollectorActor
1062                     .getFirstMatching(followerActor, AppendEntries.class);
1063
1064             assertNotNull(appendEntries);
1065
1066             assertEquals(1, appendEntries.getLeaderCommit());
1067             assertEquals(1, appendEntries.getEntries().get(0).getIndex());
1068             assertEquals(0, appendEntries.getPrevLogIndex());
1069
1070             AppendEntriesReply appendEntriesReply =
1071                 (AppendEntriesReply) MessageCollectorActor.getFirstMatching(
1072                     leaderActor, AppendEntriesReply.class);
1073
1074             assertNotNull(appendEntriesReply);
1075
1076             assertEquals(2, appendEntriesReply.getLogLastIndex());
1077             assertEquals(1, appendEntriesReply.getLogLastTerm());
1078
1079         }};
1080     }
1081
1082     @Test
1083     public void testHandleAppendEntriesReplyFailure(){
1084         new JavaTestKit(getSystem()) {
1085             {
1086
1087                 ActorRef leaderActor =
1088                     getSystem().actorOf(Props.create(MessageCollectorActor.class));
1089
1090                 ActorRef followerActor =
1091                     getSystem().actorOf(Props.create(MessageCollectorActor.class));
1092
1093
1094                 MockRaftActorContext leaderActorContext =
1095                     new MockRaftActorContext("leader", getSystem(), leaderActor);
1096
1097                 Map<String, String> peerAddresses = new HashMap<>();
1098                 peerAddresses.put("follower-1",
1099                     followerActor.path().toString());
1100
1101                 leaderActorContext.setPeerAddresses(peerAddresses);
1102
1103                 Leader leader = new Leader(leaderActorContext);
1104
1105                 AppendEntriesReply reply = new AppendEntriesReply("follower-1", 1, false, 10, 1);
1106
1107                 RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
1108
1109                 assertEquals(RaftState.Leader, raftActorBehavior.state());
1110
1111             }};
1112     }
1113
1114     @Test
1115     public void testHandleAppendEntriesReplySuccess() throws Exception {
1116         new JavaTestKit(getSystem()) {
1117             {
1118
1119                 ActorRef leaderActor =
1120                     getSystem().actorOf(Props.create(MessageCollectorActor.class));
1121
1122                 ActorRef followerActor =
1123                     getSystem().actorOf(Props.create(MessageCollectorActor.class));
1124
1125
1126                 MockRaftActorContext leaderActorContext =
1127                     new MockRaftActorContext("leader", getSystem(), leaderActor);
1128
1129                 leaderActorContext.setReplicatedLog(
1130                     new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1131
1132                 Map<String, String> peerAddresses = new HashMap<>();
1133                 peerAddresses.put("follower-1",
1134                     followerActor.path().toString());
1135
1136                 leaderActorContext.setPeerAddresses(peerAddresses);
1137                 leaderActorContext.setCommitIndex(1);
1138                 leaderActorContext.setLastApplied(1);
1139                 leaderActorContext.getTermInformation().update(1, "leader");
1140
1141                 Leader leader = new Leader(leaderActorContext);
1142
1143                 AppendEntriesReply reply = new AppendEntriesReply("follower-1", 1, true, 2, 1);
1144
1145                 RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
1146
1147                 assertEquals(RaftState.Leader, raftActorBehavior.state());
1148
1149                 assertEquals(2, leaderActorContext.getCommitIndex());
1150
1151                 ApplyLogEntries applyLogEntries =
1152                     (ApplyLogEntries) MessageCollectorActor.getFirstMatching(leaderActor,
1153                         ApplyLogEntries.class);
1154
1155                 assertNotNull(applyLogEntries);
1156
1157                 assertEquals(2, leaderActorContext.getLastApplied());
1158
1159                 assertEquals(2, applyLogEntries.getToIndex());
1160
1161                 List<Object> applyStateList = MessageCollectorActor.getAllMatching(leaderActor,
1162                     ApplyState.class);
1163
1164                 assertEquals(1,applyStateList.size());
1165
1166                 ApplyState applyState = (ApplyState) applyStateList.get(0);
1167
1168                 assertEquals(2, applyState.getReplicatedLogEntry().getIndex());
1169
1170             }};
1171     }
1172
1173     @Test
1174     public void testHandleAppendEntriesReplyUnknownFollower(){
1175         new JavaTestKit(getSystem()) {
1176             {
1177
1178                 ActorRef leaderActor =
1179                     getSystem().actorOf(Props.create(MessageCollectorActor.class));
1180
1181                 MockRaftActorContext leaderActorContext =
1182                     new MockRaftActorContext("leader", getSystem(), leaderActor);
1183
1184                 Leader leader = new Leader(leaderActorContext);
1185
1186                 AppendEntriesReply reply = new AppendEntriesReply("follower-1", 1, false, 10, 1);
1187
1188                 RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(getRef(), reply);
1189
1190                 assertEquals(RaftState.Leader, raftActorBehavior.state());
1191
1192             }};
1193     }
1194
1195     @Test
1196     public void testHandleRequestVoteReply(){
1197         new JavaTestKit(getSystem()) {
1198             {
1199
1200                 ActorRef leaderActor =
1201                     getSystem().actorOf(Props.create(MessageCollectorActor.class));
1202
1203                 MockRaftActorContext leaderActorContext =
1204                     new MockRaftActorContext("leader", getSystem(), leaderActor);
1205
1206                 Leader leader = new Leader(leaderActorContext);
1207
1208                 RaftActorBehavior raftActorBehavior = leader.handleRequestVoteReply(getRef(), new RequestVoteReply(1, true));
1209
1210                 assertEquals(RaftState.Leader, raftActorBehavior.state());
1211
1212                 raftActorBehavior = leader.handleRequestVoteReply(getRef(), new RequestVoteReply(1, false));
1213
1214                 assertEquals(RaftState.Leader, raftActorBehavior.state());
1215             }};
1216     }
1217
1218     @Test
1219     public void testIsolatedLeaderCheckNoFollowers() {
1220         new JavaTestKit(getSystem()) {{
1221             ActorRef leaderActor = getTestActor();
1222
1223             MockRaftActorContext leaderActorContext =
1224                 new MockRaftActorContext("leader", getSystem(), leaderActor);
1225
1226             Map<String, String> peerAddresses = new HashMap<>();
1227             leaderActorContext.setPeerAddresses(peerAddresses);
1228
1229             Leader leader = new Leader(leaderActorContext);
1230             RaftActorBehavior behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
1231             Assert.assertTrue(behavior instanceof Leader);
1232         }};
1233     }
1234
1235     @Test
1236     public void testIsolatedLeaderCheckTwoFollowers() throws Exception {
1237         new JavaTestKit(getSystem()) {{
1238
1239             ActorRef followerActor1 = getTestActor();
1240             ActorRef followerActor2 = getTestActor();
1241
1242             MockRaftActorContext leaderActorContext = (MockRaftActorContext) createActorContext();
1243
1244             Map<String, String> peerAddresses = new HashMap<>();
1245             peerAddresses.put("follower-1", followerActor1.path().toString());
1246             peerAddresses.put("follower-2", followerActor2.path().toString());
1247
1248             leaderActorContext.setPeerAddresses(peerAddresses);
1249
1250             Leader leader = new Leader(leaderActorContext);
1251             leader.stopIsolatedLeaderCheckSchedule();
1252
1253             leader.markFollowerActive("follower-1");
1254             leader.markFollowerActive("follower-2");
1255             RaftActorBehavior behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
1256             Assert.assertTrue("Behavior not instance of Leader when all followers are active",
1257                 behavior instanceof Leader);
1258
1259             // kill 1 follower and verify if that got killed
1260             final JavaTestKit probe = new JavaTestKit(getSystem());
1261             probe.watch(followerActor1);
1262             followerActor1.tell(PoisonPill.getInstance(), ActorRef.noSender());
1263             final Terminated termMsg1 = probe.expectMsgClass(Terminated.class);
1264             assertEquals(termMsg1.getActor(), followerActor1);
1265
1266             leader.markFollowerInActive("follower-1");
1267             leader.markFollowerActive("follower-2");
1268             behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
1269             Assert.assertTrue("Behavior not instance of Leader when majority of followers are active",
1270                 behavior instanceof Leader);
1271
1272             // kill 2nd follower and leader should change to Isolated leader
1273             followerActor2.tell(PoisonPill.getInstance(), null);
1274             probe.watch(followerActor2);
1275             followerActor2.tell(PoisonPill.getInstance(), ActorRef.noSender());
1276             final Terminated termMsg2 = probe.expectMsgClass(Terminated.class);
1277             assertEquals(termMsg2.getActor(), followerActor2);
1278
1279             leader.markFollowerInActive("follower-2");
1280             behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
1281             Assert.assertTrue("Behavior not instance of IsolatedLeader when majority followers are inactive",
1282                 behavior instanceof IsolatedLeader);
1283
1284         }};
1285     }
1286
1287
1288     @Test
1289     public void testAppendEntryCallAtEndofAppendEntryReply() throws Exception {
1290         new JavaTestKit(getSystem()) {{
1291
1292             ActorRef leaderActor = getSystem().actorOf(Props.create(MessageCollectorActor.class));
1293
1294             MockRaftActorContext leaderActorContext =
1295                 new MockRaftActorContext("leader", getSystem(), leaderActor);
1296
1297             DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
1298             configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS));
1299             configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
1300
1301             leaderActorContext.setConfigParams(configParams);
1302
1303             ActorRef followerActor = getSystem().actorOf(Props.create(ForwardMessageToBehaviorActor.class));
1304
1305             MockRaftActorContext followerActorContext =
1306                 new MockRaftActorContext("follower-reply", getSystem(), followerActor);
1307
1308             followerActorContext.setConfigParams(configParams);
1309
1310             Follower follower = new Follower(followerActorContext);
1311
1312             ForwardMessageToBehaviorActor.setBehavior(follower);
1313
1314             Map<String, String> peerAddresses = new HashMap<>();
1315             peerAddresses.put("follower-reply",
1316                 followerActor.path().toString());
1317
1318             leaderActorContext.setPeerAddresses(peerAddresses);
1319
1320             leaderActorContext.getReplicatedLog().removeFrom(0);
1321
1322             //create 3 entries
1323             leaderActorContext.setReplicatedLog(
1324                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1325
1326             leaderActorContext.setCommitIndex(1);
1327
1328             Leader leader = new Leader(leaderActorContext);
1329             leader.markFollowerActive("follower-reply");
1330
1331             Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(),
1332                 TimeUnit.MILLISECONDS);
1333
1334             leader.handleMessage(leaderActor, new SendHeartBeat());
1335
1336             AppendEntries appendEntries = (AppendEntries) ForwardMessageToBehaviorActor
1337                 .getFirstMatching(followerActor, AppendEntries.class);
1338
1339             assertNotNull(appendEntries);
1340
1341             assertEquals(1, appendEntries.getLeaderCommit());
1342             assertEquals(1, appendEntries.getEntries().get(0).getIndex());
1343             assertEquals(0, appendEntries.getPrevLogIndex());
1344
1345             AppendEntriesReply appendEntriesReply =
1346                 (AppendEntriesReply)ForwardMessageToBehaviorActor.getFirstMatching(leaderActor, AppendEntriesReply.class);
1347
1348             assertNotNull(appendEntriesReply);
1349
1350             leader.handleAppendEntriesReply(followerActor, appendEntriesReply);
1351
1352             List<Object> entries = ForwardMessageToBehaviorActor
1353                 .getAllMatching(followerActor, AppendEntries.class);
1354
1355             assertEquals("AppendEntries count should be 2 ", 2, entries.size());
1356
1357             AppendEntries appendEntriesSecond = (AppendEntries) entries.get(1);
1358
1359             assertEquals(1, appendEntriesSecond.getLeaderCommit());
1360             assertEquals(2, appendEntriesSecond.getEntries().get(0).getIndex());
1361             assertEquals(1, appendEntriesSecond.getPrevLogIndex());
1362
1363         }};
1364     }
1365
1366     class MockLeader extends Leader {
1367
1368         FollowerToSnapshot fts;
1369
1370         public MockLeader(RaftActorContext context){
1371             super(context);
1372         }
1373
1374         public FollowerToSnapshot getFollowerToSnapshot() {
1375             return fts;
1376         }
1377
1378         public void createFollowerToSnapshot(String followerId, ByteString bs ) {
1379             fts = new FollowerToSnapshot(bs);
1380             setFollowerSnapshot(followerId, fts);
1381         }
1382     }
1383
1384     private class MockConfigParamsImpl extends DefaultConfigParamsImpl {
1385
1386         private final long electionTimeOutIntervalMillis;
1387         private final int snapshotChunkSize;
1388
1389         public MockConfigParamsImpl(long electionTimeOutIntervalMillis, int snapshotChunkSize) {
1390             super();
1391             this.electionTimeOutIntervalMillis = electionTimeOutIntervalMillis;
1392             this.snapshotChunkSize = snapshotChunkSize;
1393         }
1394
1395         @Override
1396         public FiniteDuration getElectionTimeOutInterval() {
1397             return new FiniteDuration(electionTimeOutIntervalMillis, TimeUnit.MILLISECONDS);
1398         }
1399
1400         @Override
1401         public int getSnapshotChunkSize() {
1402             return snapshotChunkSize;
1403         }
1404     }
1405 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.