Add RaftActor integration tests
[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.collect.ImmutableMap;
14 import com.google.common.util.concurrent.Uninterruptibles;
15 import com.google.protobuf.ByteString;
16 import java.util.HashMap;
17 import java.util.List;
18 import java.util.Map;
19 import java.util.concurrent.TimeUnit;
20 import org.junit.After;
21 import org.junit.Assert;
22 import org.junit.Test;
23 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
24 import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
25 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
26 import org.opendaylight.controller.cluster.raft.RaftActorContext;
27 import org.opendaylight.controller.cluster.raft.RaftState;
28 import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
29 import org.opendaylight.controller.cluster.raft.SerializationUtils;
30 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
31 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
32 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
33 import org.opendaylight.controller.cluster.raft.base.messages.IsolatedLeaderCheck;
34 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
35 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
36 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
37 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader.FollowerToSnapshot;
38 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
39 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
40 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
41 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
42 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
43 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
44 import org.opendaylight.controller.cluster.raft.utils.ForwardMessageToBehaviorActor;
45 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
46 import scala.concurrent.duration.FiniteDuration;
47
48 public class LeaderTest extends AbstractLeaderTest {
49
50     static final String FOLLOWER_ID = "follower";
51
52     private final TestActorRef<ForwardMessageToBehaviorActor> leaderActor = actorFactory.createTestActor(
53             Props.create(ForwardMessageToBehaviorActor.class), actorFactory.generateActorId("leader"));
54
55     private final TestActorRef<ForwardMessageToBehaviorActor> followerActor = actorFactory.createTestActor(
56             Props.create(ForwardMessageToBehaviorActor.class), actorFactory.generateActorId("follower"));
57
58     private Leader leader;
59
60     @Override
61     @After
62     public void tearDown() throws Exception {
63         if(leader != null) {
64             leader.close();
65         }
66
67         super.tearDown();
68     }
69
70     @Test
71     public void testHandleMessageForUnknownMessage() throws Exception {
72         logStart("testHandleMessageForUnknownMessage");
73
74         leader = new Leader(createActorContext());
75
76         // handle message should return the Leader state when it receives an
77         // unknown message
78         RaftActorBehavior behavior = leader.handleMessage(followerActor, "foo");
79         Assert.assertTrue(behavior instanceof Leader);
80     }
81
82     @Test
83     public void testThatLeaderSendsAHeartbeatMessageToAllFollowers() throws Exception {
84         logStart("testThatLeaderSendsAHeartbeatMessageToAllFollowers");
85
86         MockRaftActorContext actorContext = createActorContextWithFollower();
87
88         long term = 1;
89         actorContext.getTermInformation().update(term, "");
90
91         leader = new Leader(actorContext);
92
93         // Leader should send an immediate heartbeat with no entries as follower is inactive.
94         long lastIndex = actorContext.getReplicatedLog().lastIndex();
95         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
96         assertEquals("getTerm", term, appendEntries.getTerm());
97         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
98         assertEquals("getPrevLogTerm", -1, appendEntries.getPrevLogTerm());
99         assertEquals("Entries size", 0, appendEntries.getEntries().size());
100
101         // The follower would normally reply - simulate that explicitly here.
102         leader.handleMessage(followerActor, new AppendEntriesReply(
103                 FOLLOWER_ID, term, true, lastIndex - 1, term));
104         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
105
106         followerActor.underlyingActor().clear();
107
108         // Sleep for the heartbeat interval so AppendEntries is sent.
109         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().
110                 getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS);
111
112         leader.handleMessage(leaderActor, new SendHeartBeat());
113
114         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
115         assertEquals("getPrevLogIndex", lastIndex - 1, appendEntries.getPrevLogIndex());
116         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
117         assertEquals("Entries size", 1, appendEntries.getEntries().size());
118         assertEquals("Entry getIndex", lastIndex, appendEntries.getEntries().get(0).getIndex());
119         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
120     }
121
122     @Test
123     public void testHandleReplicateMessageSendAppendEntriesToFollower() throws Exception {
124         logStart("testHandleReplicateMessageSendAppendEntriesToFollower");
125
126         MockRaftActorContext actorContext = createActorContextWithFollower();
127
128         long term = 1;
129         actorContext.getTermInformation().update(term, "");
130
131         leader = new Leader(actorContext);
132
133         // Leader will send an immediate heartbeat - ignore it.
134         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
135
136         // The follower would normally reply - simulate that explicitly here.
137         long lastIndex = actorContext.getReplicatedLog().lastIndex();
138         leader.handleMessage(followerActor, new AppendEntriesReply(
139                 FOLLOWER_ID, term, true, lastIndex, term));
140         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
141
142         followerActor.underlyingActor().clear();
143
144         MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("foo");
145         MockRaftActorContext.MockReplicatedLogEntry newEntry = new MockRaftActorContext.MockReplicatedLogEntry(
146                 1, lastIndex + 1, payload);
147         actorContext.getReplicatedLog().append(newEntry);
148         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor,
149                 new Replicate(null, null, newEntry));
150
151         // State should not change
152         assertTrue(raftBehavior instanceof Leader);
153
154         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
155         assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex());
156         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
157         assertEquals("Entries size", 1, appendEntries.getEntries().size());
158         assertEquals("Entry getIndex", lastIndex + 1, appendEntries.getEntries().get(0).getIndex());
159         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
160         assertEquals("Entry payload", payload, appendEntries.getEntries().get(0).getData());
161     }
162
163     @Test
164     public void testHandleReplicateMessageWhenThereAreNoFollowers() throws Exception {
165         logStart("testHandleReplicateMessageWhenThereAreNoFollowers");
166
167         MockRaftActorContext actorContext = createActorContext();
168
169         leader = new Leader(actorContext);
170
171         actorContext.setLastApplied(0);
172
173         long newLogIndex = actorContext.getReplicatedLog().lastIndex() + 1;
174         long term = actorContext.getTermInformation().getCurrentTerm();
175         MockRaftActorContext.MockReplicatedLogEntry newEntry = new MockRaftActorContext.MockReplicatedLogEntry(
176                 term, newLogIndex, new MockRaftActorContext.MockPayload("foo"));
177
178         actorContext.getReplicatedLog().append(newEntry);
179
180         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor,
181                 new Replicate(leaderActor, "state-id", newEntry));
182
183         // State should not change
184         assertTrue(raftBehavior instanceof Leader);
185
186         assertEquals("getCommitIndex", newLogIndex, actorContext.getCommitIndex());
187
188         // We should get 2 ApplyState messages - 1 for new log entry and 1 for the previous
189         // one since lastApplied state is 0.
190         List<ApplyState> applyStateList = MessageCollectorActor.getAllMatching(
191                 leaderActor, ApplyState.class);
192         assertEquals("ApplyState count", newLogIndex, applyStateList.size());
193
194         for(int i = 0; i <= newLogIndex - 1; i++ ) {
195             ApplyState applyState = applyStateList.get(i);
196             assertEquals("getIndex", i + 1, applyState.getReplicatedLogEntry().getIndex());
197             assertEquals("getTerm", term, applyState.getReplicatedLogEntry().getTerm());
198         }
199
200         ApplyState last = applyStateList.get((int) newLogIndex - 1);
201         assertEquals("getData", newEntry.getData(), last.getReplicatedLogEntry().getData());
202         assertEquals("getIdentifier", "state-id", last.getIdentifier());
203     }
204
205     @Test
206     public void testSendAppendEntriesOnAnInProgressInstallSnapshot() throws Exception {
207         logStart("testSendAppendEntriesOnAnInProgressInstallSnapshot");
208
209         MockRaftActorContext actorContext = createActorContextWithFollower();
210
211         Map<String, String> leadersSnapshot = new HashMap<>();
212         leadersSnapshot.put("1", "A");
213         leadersSnapshot.put("2", "B");
214         leadersSnapshot.put("3", "C");
215
216         //clears leaders log
217         actorContext.getReplicatedLog().removeFrom(0);
218
219         final int followersLastIndex = 2;
220         final int snapshotIndex = 3;
221         final int newEntryIndex = 4;
222         final int snapshotTerm = 1;
223         final int currentTerm = 2;
224
225         // set the snapshot variables in replicatedlog
226         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
227         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
228         actorContext.setCommitIndex(followersLastIndex);
229         //set follower timeout to 2 mins, helps during debugging
230         actorContext.setConfigParams(new MockConfigParamsImpl(120000L, 10));
231
232         leader = new Leader(actorContext);
233
234         // new entry
235         ReplicatedLogImplEntry entry =
236                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
237                         new MockRaftActorContext.MockPayload("D"));
238
239         //update follower timestamp
240         leader.markFollowerActive(FOLLOWER_ID);
241
242         ByteString bs = toByteString(leadersSnapshot);
243         leader.setSnapshot(Optional.of(bs));
244         FollowerToSnapshot fts = leader.new FollowerToSnapshot(bs);
245         leader.setFollowerSnapshot(FOLLOWER_ID, fts);
246
247         //send first chunk and no InstallSnapshotReply received yet
248         fts.getNextChunk();
249         fts.incrementChunkIndex();
250
251         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
252                 TimeUnit.MILLISECONDS);
253
254         leader.handleMessage(leaderActor, new SendHeartBeat());
255
256         AppendEntries aeproto = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
257
258         AppendEntries ae = (AppendEntries) SerializationUtils.fromSerializable(aeproto);
259
260         assertTrue("AppendEntries should be sent with empty entries", ae.getEntries().isEmpty());
261
262         //InstallSnapshotReply received
263         fts.markSendStatus(true);
264
265         leader.handleMessage(leaderActor, new SendHeartBeat());
266
267         InstallSnapshot is = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
268
269         assertEquals(snapshotIndex, is.getLastIncludedIndex());
270     }
271
272     @Test
273     public void testSendAppendEntriesSnapshotScenario() throws Exception {
274         logStart("testSendAppendEntriesSnapshotScenario");
275
276         MockRaftActorContext actorContext = createActorContextWithFollower();
277
278         Map<String, String> leadersSnapshot = new HashMap<>();
279         leadersSnapshot.put("1", "A");
280         leadersSnapshot.put("2", "B");
281         leadersSnapshot.put("3", "C");
282
283         //clears leaders log
284         actorContext.getReplicatedLog().removeFrom(0);
285
286         final int followersLastIndex = 2;
287         final int snapshotIndex = 3;
288         final int newEntryIndex = 4;
289         final int snapshotTerm = 1;
290         final int currentTerm = 2;
291
292         // set the snapshot variables in replicatedlog
293         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
294         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
295         actorContext.setCommitIndex(followersLastIndex);
296
297         leader = new Leader(actorContext);
298
299         // Leader will send an immediate heartbeat - ignore it.
300         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
301
302         // new entry
303         ReplicatedLogImplEntry entry =
304                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
305                         new MockRaftActorContext.MockPayload("D"));
306
307         //update follower timestamp
308         leader.markFollowerActive(FOLLOWER_ID);
309
310         // this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex
311         RaftActorBehavior raftBehavior = leader.handleMessage(
312                 leaderActor, new Replicate(null, "state-id", entry));
313
314         assertTrue(raftBehavior instanceof Leader);
315
316         MessageCollectorActor.expectFirstMatching(leaderActor, CaptureSnapshot.class);
317     }
318
319     @Test
320     public void testInitiateInstallSnapshot() throws Exception {
321         logStart("testInitiateInstallSnapshot");
322
323         MockRaftActorContext actorContext = createActorContextWithFollower();
324
325         Map<String, String> leadersSnapshot = new HashMap<>();
326         leadersSnapshot.put("1", "A");
327         leadersSnapshot.put("2", "B");
328         leadersSnapshot.put("3", "C");
329
330         //clears leaders log
331         actorContext.getReplicatedLog().removeFrom(0);
332
333         final int followersLastIndex = 2;
334         final int snapshotIndex = 3;
335         final int newEntryIndex = 4;
336         final int snapshotTerm = 1;
337         final int currentTerm = 2;
338
339         // set the snapshot variables in replicatedlog
340         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
341         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
342         actorContext.setLastApplied(3);
343         actorContext.setCommitIndex(followersLastIndex);
344
345         leader = new Leader(actorContext);
346
347         // Leader will send an immediate heartbeat - ignore it.
348         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
349
350         // set the snapshot as absent and check if capture-snapshot is invoked.
351         leader.setSnapshot(Optional.<ByteString>absent());
352
353         // new entry
354         ReplicatedLogImplEntry entry = new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
355                 new MockRaftActorContext.MockPayload("D"));
356
357         actorContext.getReplicatedLog().append(entry);
358
359         //update follower timestamp
360         leader.markFollowerActive(FOLLOWER_ID);
361
362         leader.handleMessage(leaderActor, new Replicate(null, "state-id", entry));
363
364         CaptureSnapshot cs = MessageCollectorActor.expectFirstMatching(leaderActor, CaptureSnapshot.class);
365
366         assertTrue(cs.isInstallSnapshotInitiated());
367         assertEquals(3, cs.getLastAppliedIndex());
368         assertEquals(1, cs.getLastAppliedTerm());
369         assertEquals(4, cs.getLastIndex());
370         assertEquals(2, cs.getLastTerm());
371
372         // if an initiate is started again when first is in progress, it shouldnt initiate Capture
373         leader.handleMessage(leaderActor, new Replicate(null, "state-id", entry));
374
375         List<CaptureSnapshot> captureSnapshots = MessageCollectorActor.getAllMatching(leaderActor, CaptureSnapshot.class);
376         assertEquals("CaptureSnapshot should not get invoked when  initiate is in progress", 1, captureSnapshots.size());
377     }
378
379     @Test
380     public void testInstallSnapshot() throws Exception {
381         logStart("testInstallSnapshot");
382
383         MockRaftActorContext actorContext = createActorContextWithFollower();
384
385         Map<String, String> leadersSnapshot = new HashMap<>();
386         leadersSnapshot.put("1", "A");
387         leadersSnapshot.put("2", "B");
388         leadersSnapshot.put("3", "C");
389
390         //clears leaders log
391         actorContext.getReplicatedLog().removeFrom(0);
392
393         final int followersLastIndex = 2;
394         final int snapshotIndex = 3;
395         final int snapshotTerm = 1;
396         final int currentTerm = 2;
397
398         // set the snapshot variables in replicatedlog
399         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
400         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
401         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
402         actorContext.setCommitIndex(followersLastIndex);
403
404         leader = new Leader(actorContext);
405
406         // Ignore initial heartbeat.
407         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
408
409         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor,
410                 new SendInstallSnapshot(toByteString(leadersSnapshot)));
411
412         assertTrue(raftBehavior instanceof Leader);
413
414         // check if installsnapshot gets called with the correct values.
415
416         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
417
418         assertNotNull(installSnapshot.getData());
419         assertEquals(snapshotIndex, installSnapshot.getLastIncludedIndex());
420         assertEquals(snapshotTerm, installSnapshot.getLastIncludedTerm());
421
422         assertEquals(currentTerm, installSnapshot.getTerm());
423     }
424
425     @Test
426     public void testHandleInstallSnapshotReplyLastChunk() throws Exception {
427         logStart("testHandleInstallSnapshotReplyLastChunk");
428
429         MockRaftActorContext actorContext = createActorContextWithFollower();
430
431         final int followersLastIndex = 2;
432         final int snapshotIndex = 3;
433         final int snapshotTerm = 1;
434         final int currentTerm = 2;
435
436         actorContext.setCommitIndex(followersLastIndex);
437
438         leader = new Leader(actorContext);
439
440         // Ignore initial heartbeat.
441         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
442
443         Map<String, String> leadersSnapshot = new HashMap<>();
444         leadersSnapshot.put("1", "A");
445         leadersSnapshot.put("2", "B");
446         leadersSnapshot.put("3", "C");
447
448         // set the snapshot variables in replicatedlog
449
450         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
451         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
452         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
453
454         ByteString bs = toByteString(leadersSnapshot);
455         leader.setSnapshot(Optional.of(bs));
456         FollowerToSnapshot fts = leader.new FollowerToSnapshot(bs);
457         leader.setFollowerSnapshot(FOLLOWER_ID, fts);
458         while(!fts.isLastChunk(fts.getChunkIndex())) {
459             fts.getNextChunk();
460             fts.incrementChunkIndex();
461         }
462
463         //clears leaders log
464         actorContext.getReplicatedLog().removeFrom(0);
465
466         RaftActorBehavior raftBehavior = leader.handleMessage(followerActor,
467                 new InstallSnapshotReply(currentTerm, FOLLOWER_ID, fts.getChunkIndex(), true));
468
469         assertTrue(raftBehavior instanceof Leader);
470
471         assertEquals(0, leader.followerSnapshotSize());
472         assertEquals(1, leader.followerLogSize());
473         FollowerLogInformation fli = leader.getFollower(FOLLOWER_ID);
474         assertNotNull(fli);
475         assertEquals(snapshotIndex, fli.getMatchIndex());
476         assertEquals(snapshotIndex, fli.getMatchIndex());
477         assertEquals(snapshotIndex + 1, fli.getNextIndex());
478     }
479
480     @Test
481     public void testSendSnapshotfromInstallSnapshotReply() throws Exception {
482         logStart("testSendSnapshotfromInstallSnapshotReply");
483
484         MockRaftActorContext actorContext = createActorContextWithFollower();
485
486         final int followersLastIndex = 2;
487         final int snapshotIndex = 3;
488         final int snapshotTerm = 1;
489         final int currentTerm = 2;
490
491         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(){
492             @Override
493             public int getSnapshotChunkSize() {
494                 return 50;
495             }
496         };
497         configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS));
498         configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
499
500         actorContext.setConfigParams(configParams);
501         actorContext.setCommitIndex(followersLastIndex);
502
503         leader = new Leader(actorContext);
504
505         Map<String, String> leadersSnapshot = new HashMap<>();
506         leadersSnapshot.put("1", "A");
507         leadersSnapshot.put("2", "B");
508         leadersSnapshot.put("3", "C");
509
510         // set the snapshot variables in replicatedlog
511         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
512         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
513         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
514
515         ByteString bs = toByteString(leadersSnapshot);
516         leader.setSnapshot(Optional.of(bs));
517
518         leader.handleMessage(leaderActor, new SendInstallSnapshot(bs));
519
520         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
521
522         assertEquals(1, installSnapshot.getChunkIndex());
523         assertEquals(3, installSnapshot.getTotalChunks());
524
525         followerActor.underlyingActor().clear();
526         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
527                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
528
529         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
530
531         assertEquals(2, installSnapshot.getChunkIndex());
532         assertEquals(3, installSnapshot.getTotalChunks());
533
534         followerActor.underlyingActor().clear();
535         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
536                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
537
538         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
539
540         // Send snapshot reply one more time and make sure that a new snapshot message should not be sent to follower
541         followerActor.underlyingActor().clear();
542         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
543                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
544
545         installSnapshot = MessageCollectorActor.getFirstMatching(followerActor, InstallSnapshot.class);
546
547         Assert.assertNull(installSnapshot);
548     }
549
550
551     @Test
552     public void testHandleInstallSnapshotReplyWithInvalidChunkIndex() throws Exception{
553         logStart("testHandleInstallSnapshotReplyWithInvalidChunkIndex");
554
555         MockRaftActorContext actorContext = createActorContextWithFollower();
556
557         final int followersLastIndex = 2;
558         final int snapshotIndex = 3;
559         final int snapshotTerm = 1;
560         final int currentTerm = 2;
561
562         actorContext.setConfigParams(new DefaultConfigParamsImpl(){
563             @Override
564             public int getSnapshotChunkSize() {
565                 return 50;
566             }
567         });
568
569         actorContext.setCommitIndex(followersLastIndex);
570
571         leader = new Leader(actorContext);
572
573         Map<String, String> leadersSnapshot = new HashMap<>();
574         leadersSnapshot.put("1", "A");
575         leadersSnapshot.put("2", "B");
576         leadersSnapshot.put("3", "C");
577
578         // set the snapshot variables in replicatedlog
579         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
580         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
581         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
582
583         ByteString bs = toByteString(leadersSnapshot);
584         leader.setSnapshot(Optional.of(bs));
585
586         Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
587         leader.handleMessage(leaderActor, new SendInstallSnapshot(bs));
588
589         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
590
591         assertEquals(1, installSnapshot.getChunkIndex());
592         assertEquals(3, installSnapshot.getTotalChunks());
593
594         followerActor.underlyingActor().clear();
595
596         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
597                 FOLLOWER_ID, -1, false));
598
599         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
600                 TimeUnit.MILLISECONDS);
601
602         leader.handleMessage(leaderActor, new SendHeartBeat());
603
604         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
605
606         assertEquals(1, installSnapshot.getChunkIndex());
607         assertEquals(3, installSnapshot.getTotalChunks());
608     }
609
610     @Test
611     public void testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk() throws Exception {
612         logStart("testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk");
613
614         MockRaftActorContext actorContext = createActorContextWithFollower();
615
616         final int followersLastIndex = 2;
617         final int snapshotIndex = 3;
618         final int snapshotTerm = 1;
619         final int currentTerm = 2;
620
621         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
622             @Override
623             public int getSnapshotChunkSize() {
624                 return 50;
625             }
626         });
627
628         actorContext.setCommitIndex(followersLastIndex);
629
630         leader = new Leader(actorContext);
631
632         Map<String, String> leadersSnapshot = new HashMap<>();
633         leadersSnapshot.put("1", "A");
634         leadersSnapshot.put("2", "B");
635         leadersSnapshot.put("3", "C");
636
637         // set the snapshot variables in replicatedlog
638         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
639         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
640         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
641
642         ByteString bs = toByteString(leadersSnapshot);
643         leader.setSnapshot(Optional.of(bs));
644
645         leader.handleMessage(leaderActor, new SendInstallSnapshot(bs));
646
647         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
648
649         assertEquals(1, installSnapshot.getChunkIndex());
650         assertEquals(3, installSnapshot.getTotalChunks());
651         assertEquals(AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE, installSnapshot.getLastChunkHashCode().get().intValue());
652
653         int hashCode = installSnapshot.getData().hashCode();
654
655         followerActor.underlyingActor().clear();
656
657         leader.handleMessage(followerActor, new InstallSnapshotReply(installSnapshot.getTerm(),
658                 FOLLOWER_ID, 1, true));
659
660         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
661
662         assertEquals(2, installSnapshot.getChunkIndex());
663         assertEquals(3, installSnapshot.getTotalChunks());
664         assertEquals(hashCode, installSnapshot.getLastChunkHashCode().get().intValue());
665     }
666
667     @Test
668     public void testFollowerToSnapshotLogic() {
669         logStart("testFollowerToSnapshotLogic");
670
671         MockRaftActorContext actorContext = createActorContext();
672
673         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
674             @Override
675             public int getSnapshotChunkSize() {
676                 return 50;
677             }
678         });
679
680         leader = new Leader(actorContext);
681
682         Map<String, String> leadersSnapshot = new HashMap<>();
683         leadersSnapshot.put("1", "A");
684         leadersSnapshot.put("2", "B");
685         leadersSnapshot.put("3", "C");
686
687         ByteString bs = toByteString(leadersSnapshot);
688         byte[] barray = bs.toByteArray();
689
690         FollowerToSnapshot fts = leader.new FollowerToSnapshot(bs);
691         leader.setFollowerSnapshot(FOLLOWER_ID, fts);
692
693         assertEquals(bs.size(), barray.length);
694
695         int chunkIndex=0;
696         for (int i=0; i < barray.length; i = i + 50) {
697             int j = i + 50;
698             chunkIndex++;
699
700             if (i + 50 > barray.length) {
701                 j = barray.length;
702             }
703
704             ByteString chunk = fts.getNextChunk();
705             assertEquals("bytestring size not matching for chunk:"+ chunkIndex, j-i, chunk.size());
706             assertEquals("chunkindex not matching", chunkIndex, fts.getChunkIndex());
707
708             fts.markSendStatus(true);
709             if (!fts.isLastChunk(chunkIndex)) {
710                 fts.incrementChunkIndex();
711             }
712         }
713
714         assertEquals("totalChunks not matching", chunkIndex, fts.getTotalChunks());
715     }
716
717     @Override protected RaftActorBehavior createBehavior(
718         RaftActorContext actorContext) {
719         return new Leader(actorContext);
720     }
721
722     @Override
723     protected MockRaftActorContext createActorContext() {
724         return createActorContext(leaderActor);
725     }
726
727     @Override
728     protected MockRaftActorContext createActorContext(ActorRef actorRef) {
729         return createActorContext("leader", actorRef);
730     }
731
732     private MockRaftActorContext createActorContextWithFollower() {
733         MockRaftActorContext actorContext = createActorContext();
734         actorContext.setPeerAddresses(ImmutableMap.<String, String>builder().put(FOLLOWER_ID,
735                 followerActor.path().toString()).build());
736         return actorContext;
737     }
738
739     private MockRaftActorContext createActorContext(String id, ActorRef actorRef) {
740         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
741         configParams.setHeartBeatInterval(new FiniteDuration(50, TimeUnit.MILLISECONDS));
742         configParams.setElectionTimeoutFactor(100000);
743         MockRaftActorContext context = new MockRaftActorContext(id, getSystem(), actorRef);
744         context.setConfigParams(configParams);
745         return context;
746     }
747
748     @Test
749     public void testLeaderCreatedWithCommitIndexLessThanLastIndex() throws Exception {
750         logStart("testLeaderCreatedWithCommitIndexLessThanLastIndex");
751
752         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
753
754         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
755
756         Follower follower = new Follower(followerActorContext);
757         followerActor.underlyingActor().setBehavior(follower);
758
759         Map<String, String> peerAddresses = new HashMap<>();
760         peerAddresses.put(FOLLOWER_ID, followerActor.path().toString());
761
762         leaderActorContext.setPeerAddresses(peerAddresses);
763
764         leaderActorContext.getReplicatedLog().removeFrom(0);
765
766         //create 3 entries
767         leaderActorContext.setReplicatedLog(
768                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
769
770         leaderActorContext.setCommitIndex(1);
771
772         followerActorContext.getReplicatedLog().removeFrom(0);
773
774         // follower too has the exact same log entries and has the same commit index
775         followerActorContext.setReplicatedLog(
776                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
777
778         followerActorContext.setCommitIndex(1);
779
780         leader = new Leader(leaderActorContext);
781
782         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
783
784         assertEquals(1, appendEntries.getLeaderCommit());
785         assertEquals(0, appendEntries.getEntries().size());
786         assertEquals(0, appendEntries.getPrevLogIndex());
787
788         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(
789                 leaderActor, AppendEntriesReply.class);
790
791         assertEquals(2, appendEntriesReply.getLogLastIndex());
792         assertEquals(1, appendEntriesReply.getLogLastTerm());
793
794         // follower returns its next index
795         assertEquals(2, appendEntriesReply.getLogLastIndex());
796         assertEquals(1, appendEntriesReply.getLogLastTerm());
797
798         follower.close();
799     }
800
801     @Test
802     public void testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex() throws Exception {
803         logStart("testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex");
804
805         MockRaftActorContext leaderActorContext = createActorContext();
806
807         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
808
809         Follower follower = new Follower(followerActorContext);
810         followerActor.underlyingActor().setBehavior(follower);
811
812         Map<String, String> peerAddresses = new HashMap<>();
813         peerAddresses.put(FOLLOWER_ID, followerActor.path().toString());
814
815         leaderActorContext.setPeerAddresses(peerAddresses);
816
817         leaderActorContext.getReplicatedLog().removeFrom(0);
818
819         leaderActorContext.setReplicatedLog(
820                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
821
822         leaderActorContext.setCommitIndex(1);
823
824         followerActorContext.getReplicatedLog().removeFrom(0);
825
826         followerActorContext.setReplicatedLog(
827                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
828
829         // follower has the same log entries but its commit index > leaders commit index
830         followerActorContext.setCommitIndex(2);
831
832         leader = new Leader(leaderActorContext);
833
834         // Initial heartbeat
835         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
836
837         assertEquals(1, appendEntries.getLeaderCommit());
838         assertEquals(0, appendEntries.getEntries().size());
839         assertEquals(0, appendEntries.getPrevLogIndex());
840
841         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(
842                 leaderActor, AppendEntriesReply.class);
843
844         assertEquals(2, appendEntriesReply.getLogLastIndex());
845         assertEquals(1, appendEntriesReply.getLogLastTerm());
846
847         leaderActor.underlyingActor().setBehavior(follower);
848         leader.handleMessage(followerActor, appendEntriesReply);
849
850         leaderActor.underlyingActor().clear();
851         followerActor.underlyingActor().clear();
852
853         Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(),
854                 TimeUnit.MILLISECONDS);
855
856         leader.handleMessage(leaderActor, new SendHeartBeat());
857
858         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
859
860         assertEquals(2, appendEntries.getLeaderCommit());
861         assertEquals(0, appendEntries.getEntries().size());
862         assertEquals(2, appendEntries.getPrevLogIndex());
863
864         appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
865
866         assertEquals(2, appendEntriesReply.getLogLastIndex());
867         assertEquals(1, appendEntriesReply.getLogLastTerm());
868
869         assertEquals(2, followerActorContext.getCommitIndex());
870
871         follower.close();
872     }
873
874     @Test
875     public void testHandleAppendEntriesReplyFailure(){
876         logStart("testHandleAppendEntriesReplyFailure");
877
878         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
879
880         leader = new Leader(leaderActorContext);
881
882         // Send initial heartbeat reply with last index.
883         leader.handleAppendEntriesReply(followerActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 10, 1));
884
885         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
886         assertEquals("getNextIndex", 11, followerInfo.getNextIndex());
887
888         AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, false, 10, 1);
889
890         RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
891
892         assertEquals(RaftState.Leader, raftActorBehavior.state());
893
894         assertEquals("getNextIndex", 10, followerInfo.getNextIndex());
895     }
896
897     @Test
898     public void testHandleAppendEntriesReplySuccess() throws Exception {
899         logStart("testHandleAppendEntriesReplySuccess");
900
901         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
902
903         leaderActorContext.setReplicatedLog(
904                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
905
906         leaderActorContext.setCommitIndex(1);
907         leaderActorContext.setLastApplied(1);
908         leaderActorContext.getTermInformation().update(1, "leader");
909
910         leader = new Leader(leaderActorContext);
911
912         AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, true, 2, 1);
913
914         RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
915
916         assertEquals(RaftState.Leader, raftActorBehavior.state());
917
918         assertEquals(2, leaderActorContext.getCommitIndex());
919
920         ApplyJournalEntries applyJournalEntries = MessageCollectorActor.expectFirstMatching(
921                 leaderActor, ApplyJournalEntries.class);
922
923         assertEquals(2, leaderActorContext.getLastApplied());
924
925         assertEquals(2, applyJournalEntries.getToIndex());
926
927         List<ApplyState> applyStateList = MessageCollectorActor.getAllMatching(leaderActor,
928                 ApplyState.class);
929
930         assertEquals(1,applyStateList.size());
931
932         ApplyState applyState = applyStateList.get(0);
933
934         assertEquals(2, applyState.getReplicatedLogEntry().getIndex());
935     }
936
937     @Test
938     public void testHandleAppendEntriesReplyUnknownFollower(){
939         logStart("testHandleAppendEntriesReplyUnknownFollower");
940
941         MockRaftActorContext leaderActorContext = createActorContext();
942
943         leader = new Leader(leaderActorContext);
944
945         AppendEntriesReply reply = new AppendEntriesReply("unkown-follower", 1, false, 10, 1);
946
947         RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
948
949         assertEquals(RaftState.Leader, raftActorBehavior.state());
950     }
951
952     @Test
953     public void testHandleRequestVoteReply(){
954         logStart("testHandleRequestVoteReply");
955
956         MockRaftActorContext leaderActorContext = createActorContext();
957
958         leader = new Leader(leaderActorContext);
959
960         // Should be a no-op.
961         RaftActorBehavior raftActorBehavior = leader.handleRequestVoteReply(followerActor,
962                 new RequestVoteReply(1, true));
963
964         assertEquals(RaftState.Leader, raftActorBehavior.state());
965
966         raftActorBehavior = leader.handleRequestVoteReply(followerActor, new RequestVoteReply(1, false));
967
968         assertEquals(RaftState.Leader, raftActorBehavior.state());
969     }
970
971     @Test
972     public void testIsolatedLeaderCheckNoFollowers() {
973         logStart("testIsolatedLeaderCheckNoFollowers");
974
975         MockRaftActorContext leaderActorContext = createActorContext();
976
977         leader = new Leader(leaderActorContext);
978         RaftActorBehavior behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
979         Assert.assertTrue(behavior instanceof Leader);
980     }
981
982     @Test
983     public void testIsolatedLeaderCheckTwoFollowers() throws Exception {
984         logStart("testIsolatedLeaderCheckTwoFollowers");
985
986         new JavaTestKit(getSystem()) {{
987
988             ActorRef followerActor1 = getTestActor();
989             ActorRef followerActor2 = getTestActor();
990
991             MockRaftActorContext leaderActorContext = createActorContext();
992
993             Map<String, String> peerAddresses = new HashMap<>();
994             peerAddresses.put("follower-1", followerActor1.path().toString());
995             peerAddresses.put("follower-2", followerActor2.path().toString());
996
997             leaderActorContext.setPeerAddresses(peerAddresses);
998
999             leader = new Leader(leaderActorContext);
1000
1001             leader.markFollowerActive("follower-1");
1002             leader.markFollowerActive("follower-2");
1003             RaftActorBehavior behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
1004             Assert.assertTrue("Behavior not instance of Leader when all followers are active",
1005                 behavior instanceof Leader);
1006
1007             // kill 1 follower and verify if that got killed
1008             final JavaTestKit probe = new JavaTestKit(getSystem());
1009             probe.watch(followerActor1);
1010             followerActor1.tell(PoisonPill.getInstance(), ActorRef.noSender());
1011             final Terminated termMsg1 = probe.expectMsgClass(Terminated.class);
1012             assertEquals(termMsg1.getActor(), followerActor1);
1013
1014             leader.markFollowerInActive("follower-1");
1015             leader.markFollowerActive("follower-2");
1016             behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
1017             Assert.assertTrue("Behavior not instance of Leader when majority of followers are active",
1018                 behavior instanceof Leader);
1019
1020             // kill 2nd follower and leader should change to Isolated leader
1021             followerActor2.tell(PoisonPill.getInstance(), null);
1022             probe.watch(followerActor2);
1023             followerActor2.tell(PoisonPill.getInstance(), ActorRef.noSender());
1024             final Terminated termMsg2 = probe.expectMsgClass(Terminated.class);
1025             assertEquals(termMsg2.getActor(), followerActor2);
1026
1027             leader.markFollowerInActive("follower-2");
1028             behavior = leader.handleMessage(leaderActor, new IsolatedLeaderCheck());
1029             Assert.assertTrue("Behavior not instance of IsolatedLeader when majority followers are inactive",
1030                 behavior instanceof IsolatedLeader);
1031         }};
1032     }
1033
1034
1035     @Test
1036     public void testAppendEntryCallAtEndofAppendEntryReply() throws Exception {
1037         logStart("testAppendEntryCallAtEndofAppendEntryReply");
1038
1039         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1040
1041         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
1042         //configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS));
1043         configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
1044
1045         leaderActorContext.setConfigParams(configParams);
1046
1047         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
1048
1049         followerActorContext.setConfigParams(configParams);
1050
1051         Follower follower = new Follower(followerActorContext);
1052         followerActor.underlyingActor().setBehavior(follower);
1053
1054         leaderActorContext.getReplicatedLog().removeFrom(0);
1055         leaderActorContext.setCommitIndex(-1);
1056         leaderActorContext.setLastApplied(-1);
1057
1058         followerActorContext.getReplicatedLog().removeFrom(0);
1059         followerActorContext.setCommitIndex(-1);
1060         followerActorContext.setLastApplied(-1);
1061
1062         leader = new Leader(leaderActorContext);
1063
1064         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(
1065                 leaderActor, AppendEntriesReply.class);
1066
1067         leader.handleMessage(followerActor, appendEntriesReply);
1068
1069         // Clear initial heartbeat messages
1070
1071         leaderActor.underlyingActor().clear();
1072         followerActor.underlyingActor().clear();
1073
1074         // create 3 entries
1075         leaderActorContext.setReplicatedLog(
1076                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1077         leaderActorContext.setCommitIndex(1);
1078         leaderActorContext.setLastApplied(1);
1079
1080         Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(),
1081                 TimeUnit.MILLISECONDS);
1082
1083         leader.handleMessage(leaderActor, new SendHeartBeat());
1084
1085         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1086
1087         // Should send first log entry
1088         assertEquals(1, appendEntries.getLeaderCommit());
1089         assertEquals(0, appendEntries.getEntries().get(0).getIndex());
1090         assertEquals(-1, appendEntries.getPrevLogIndex());
1091
1092         appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1093
1094         assertEquals(1, appendEntriesReply.getLogLastTerm());
1095         assertEquals(0, appendEntriesReply.getLogLastIndex());
1096
1097         followerActor.underlyingActor().clear();
1098
1099         leader.handleAppendEntriesReply(followerActor, appendEntriesReply);
1100
1101         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1102
1103         // Should send second log entry
1104         assertEquals(1, appendEntries.getLeaderCommit());
1105         assertEquals(1, appendEntries.getEntries().get(0).getIndex());
1106
1107         follower.close();
1108     }
1109
1110     @Test
1111     public void testLaggingFollowerStarvation() throws Exception {
1112         logStart("testLaggingFollowerStarvation");
1113         new JavaTestKit(getSystem()) {{
1114             String leaderActorId = actorFactory.generateActorId("leader");
1115             String follower1ActorId = actorFactory.generateActorId("follower");
1116             String follower2ActorId = actorFactory.generateActorId("follower");
1117
1118             TestActorRef<ForwardMessageToBehaviorActor> leaderActor =
1119                     actorFactory.createTestActor(ForwardMessageToBehaviorActor.props(), leaderActorId);
1120             ActorRef follower1Actor = actorFactory.createActor(MessageCollectorActor.props(), follower1ActorId);
1121             ActorRef follower2Actor = actorFactory.createActor(MessageCollectorActor.props(), follower2ActorId);
1122
1123             MockRaftActorContext leaderActorContext =
1124                     new MockRaftActorContext(leaderActorId, getSystem(), leaderActor);
1125
1126             DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
1127             configParams.setHeartBeatInterval(new FiniteDuration(200, TimeUnit.MILLISECONDS));
1128             configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
1129
1130             leaderActorContext.setConfigParams(configParams);
1131
1132             leaderActorContext.setReplicatedLog(
1133                     new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(1,5,1).build());
1134
1135             Map<String, String> peerAddresses = new HashMap<>();
1136             peerAddresses.put(follower1ActorId,
1137                     follower1Actor.path().toString());
1138             peerAddresses.put(follower2ActorId,
1139                     follower2Actor.path().toString());
1140
1141             leaderActorContext.setPeerAddresses(peerAddresses);
1142             leaderActorContext.getTermInformation().update(1, leaderActorId);
1143
1144             RaftActorBehavior leader = createBehavior(leaderActorContext);
1145
1146             leaderActor.underlyingActor().setBehavior(leader);
1147
1148             for(int i=1;i<6;i++) {
1149                 // Each AppendEntriesReply could end up rescheduling the heartbeat (without the fix for bug 2733)
1150                 RaftActorBehavior newBehavior = leader.handleMessage(follower1Actor, new AppendEntriesReply(follower1ActorId, 1, true, i, 1));
1151                 assertTrue(newBehavior == leader);
1152                 Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
1153             }
1154
1155             // Check if the leader has been receiving SendHeartbeat messages despite getting AppendEntriesReply
1156             List<SendHeartBeat> heartbeats = MessageCollectorActor.getAllMatching(leaderActor, SendHeartBeat.class);
1157
1158             assertTrue(String.format("%s heartbeat(s) is less than expected", heartbeats.size()),
1159                     heartbeats.size() > 1);
1160
1161             // Check if follower-2 got AppendEntries during this time and was not starved
1162             List<AppendEntries> appendEntries = MessageCollectorActor.getAllMatching(follower2Actor, AppendEntries.class);
1163
1164             assertTrue(String.format("%s append entries is less than expected", appendEntries.size()),
1165                     appendEntries.size() > 1);
1166
1167         }};
1168     }
1169
1170     @Override
1171     protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(RaftActorContext actorContext,
1172             ActorRef actorRef, RaftRPC rpc) throws Exception {
1173         super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
1174         assertEquals("New votedFor", null, actorContext.getTermInformation().getVotedFor());
1175     }
1176
1177     private class MockConfigParamsImpl extends DefaultConfigParamsImpl {
1178
1179         private final long electionTimeOutIntervalMillis;
1180         private final int snapshotChunkSize;
1181
1182         public MockConfigParamsImpl(long electionTimeOutIntervalMillis, int snapshotChunkSize) {
1183             super();
1184             this.electionTimeOutIntervalMillis = electionTimeOutIntervalMillis;
1185             this.snapshotChunkSize = snapshotChunkSize;
1186         }
1187
1188         @Override
1189         public FiniteDuration getElectionTimeOutInterval() {
1190             return new FiniteDuration(electionTimeOutIntervalMillis, TimeUnit.MILLISECONDS);
1191         }
1192
1193         @Override
1194         public int getSnapshotChunkSize() {
1195             return snapshotChunkSize;
1196         }
1197     }
1198 }

©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.