Hide IsolatedLeaderCheck message
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / behaviors / LeaderTest.java
1 /*
2  * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8
9 package org.opendaylight.controller.cluster.raft.behaviors;
10
11 import static org.junit.Assert.assertEquals;
12 import static org.junit.Assert.assertNotNull;
13 import static org.junit.Assert.assertTrue;
14 import static org.mockito.Mockito.mock;
15 import static org.mockito.Mockito.never;
16 import static org.mockito.Mockito.verify;
17 import akka.actor.ActorRef;
18 import akka.actor.PoisonPill;
19 import akka.actor.Props;
20 import akka.actor.Terminated;
21 import akka.testkit.JavaTestKit;
22 import akka.testkit.TestActorRef;
23 import com.google.common.collect.ImmutableMap;
24 import com.google.common.util.concurrent.Uninterruptibles;
25 import com.google.protobuf.ByteString;
26 import java.util.Arrays;
27 import java.util.Collections;
28 import java.util.HashMap;
29 import java.util.List;
30 import java.util.Map;
31 import java.util.concurrent.TimeUnit;
32 import org.junit.After;
33 import org.junit.Assert;
34 import org.junit.Test;
35 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
36 import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
37 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
38 import org.opendaylight.controller.cluster.raft.RaftActorContext;
39 import org.opendaylight.controller.cluster.raft.RaftActorLeadershipTransferCohort;
40 import org.opendaylight.controller.cluster.raft.RaftState;
41 import org.opendaylight.controller.cluster.raft.RaftVersions;
42 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
43 import org.opendaylight.controller.cluster.raft.ReplicatedLogImplEntry;
44 import org.opendaylight.controller.cluster.raft.SerializationUtils;
45 import org.opendaylight.controller.cluster.raft.Snapshot;
46 import org.opendaylight.controller.cluster.raft.VotingState;
47 import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries;
48 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
49 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
50 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
51 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
52 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
53 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
54 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader.FollowerToSnapshot;
55 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
56 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
57 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
58 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
59 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
60 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
61 import org.opendaylight.controller.cluster.raft.policy.DefaultRaftPolicy;
62 import org.opendaylight.controller.cluster.raft.policy.RaftPolicy;
63 import org.opendaylight.controller.cluster.raft.utils.ForwardMessageToBehaviorActor;
64 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
65 import scala.concurrent.duration.FiniteDuration;
66
67 public class LeaderTest extends AbstractLeaderTest {
68
69     static final String FOLLOWER_ID = "follower";
70     public static final String LEADER_ID = "leader";
71
72     private final TestActorRef<ForwardMessageToBehaviorActor> leaderActor = actorFactory.createTestActor(
73             Props.create(ForwardMessageToBehaviorActor.class), actorFactory.generateActorId("leader"));
74
75     private final TestActorRef<ForwardMessageToBehaviorActor> followerActor = actorFactory.createTestActor(
76             Props.create(ForwardMessageToBehaviorActor.class), actorFactory.generateActorId("follower"));
77
78     private Leader leader;
79     private final short payloadVersion = 5;
80
81     @Override
82     @After
83     public void tearDown() throws Exception {
84         if(leader != null) {
85             leader.close();
86         }
87
88         super.tearDown();
89     }
90
91     @Test
92     public void testHandleMessageForUnknownMessage() throws Exception {
93         logStart("testHandleMessageForUnknownMessage");
94
95         leader = new Leader(createActorContext());
96
97         // handle message should return the Leader state when it receives an
98         // unknown message
99         RaftActorBehavior behavior = leader.handleMessage(followerActor, "foo");
100         Assert.assertTrue(behavior instanceof Leader);
101     }
102
103     @Test
104     public void testThatLeaderSendsAHeartbeatMessageToAllFollowers() throws Exception {
105         logStart("testThatLeaderSendsAHeartbeatMessageToAllFollowers");
106
107         MockRaftActorContext actorContext = createActorContextWithFollower();
108         short payloadVersion = (short)5;
109         actorContext.setPayloadVersion(payloadVersion);
110
111         long term = 1;
112         actorContext.getTermInformation().update(term, "");
113
114         leader = new Leader(actorContext);
115
116         // Leader should send an immediate heartbeat with no entries as follower is inactive.
117         long lastIndex = actorContext.getReplicatedLog().lastIndex();
118         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
119         assertEquals("getTerm", term, appendEntries.getTerm());
120         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
121         assertEquals("getPrevLogTerm", -1, appendEntries.getPrevLogTerm());
122         assertEquals("Entries size", 0, appendEntries.getEntries().size());
123         assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion());
124
125         // The follower would normally reply - simulate that explicitly here.
126         leader.handleMessage(followerActor, new AppendEntriesReply(
127                 FOLLOWER_ID, term, true, lastIndex - 1, term, (short)0));
128         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
129
130         followerActor.underlyingActor().clear();
131
132         // Sleep for the heartbeat interval so AppendEntries is sent.
133         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().
134                 getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS);
135
136         leader.handleMessage(leaderActor, new SendHeartBeat());
137
138         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
139         assertEquals("getPrevLogIndex", lastIndex - 1, appendEntries.getPrevLogIndex());
140         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
141         assertEquals("Entries size", 1, appendEntries.getEntries().size());
142         assertEquals("Entry getIndex", lastIndex, appendEntries.getEntries().get(0).getIndex());
143         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
144         assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion());
145     }
146
147
148     private RaftActorBehavior sendReplicate(MockRaftActorContext actorContext, long index){
149         return sendReplicate(actorContext, 1, index);
150     }
151
152     private RaftActorBehavior sendReplicate(MockRaftActorContext actorContext, long term, long index){
153         MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("foo");
154         MockRaftActorContext.MockReplicatedLogEntry newEntry = new MockRaftActorContext.MockReplicatedLogEntry(
155                 term, index, payload);
156         actorContext.getReplicatedLog().append(newEntry);
157         return leader.handleMessage(leaderActor, new Replicate(null, null, newEntry));
158     }
159
160     @Test
161     public void testHandleReplicateMessageSendAppendEntriesToFollower() throws Exception {
162         logStart("testHandleReplicateMessageSendAppendEntriesToFollower");
163
164         MockRaftActorContext actorContext = createActorContextWithFollower();
165
166         long term = 1;
167         actorContext.getTermInformation().update(term, "");
168
169         leader = new Leader(actorContext);
170
171         // Leader will send an immediate heartbeat - ignore it.
172         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
173
174         // The follower would normally reply - simulate that explicitly here.
175         long lastIndex = actorContext.getReplicatedLog().lastIndex();
176         leader.handleMessage(followerActor, new AppendEntriesReply(
177                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
178         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
179
180         followerActor.underlyingActor().clear();
181
182         RaftActorBehavior raftBehavior = sendReplicate(actorContext, lastIndex + 1);
183
184         // State should not change
185         assertTrue(raftBehavior instanceof Leader);
186
187         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
188         assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex());
189         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
190         assertEquals("Entries size", 1, appendEntries.getEntries().size());
191         assertEquals("Entry getIndex", lastIndex + 1, appendEntries.getEntries().get(0).getIndex());
192         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
193         assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString());
194         assertEquals("Commit Index", lastIndex, actorContext.getCommitIndex());
195     }
196
197     @Test
198     public void testHandleReplicateMessageWithHigherTermThanPreviousEntry() throws Exception {
199         logStart("testHandleReplicateMessageWithHigherTermThanPreviousEntry");
200
201         MockRaftActorContext actorContext = createActorContextWithFollower();
202
203         // The raft context is initialized with a couple log entries. However the commitIndex
204         // is -1, simulating that the leader previously didn't get consensus and thus the log entries weren't
205         // committed and applied. Now it regains leadership with a higher term (2).
206         long prevTerm = actorContext.getTermInformation().getCurrentTerm();
207         long newTerm = prevTerm + 1;
208         actorContext.getTermInformation().update(newTerm, "");
209
210         leader = new Leader(actorContext);
211
212         // Leader will send an immediate heartbeat - ignore it.
213         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
214
215         // The follower replies with the leader's current last index and term, simulating that it is
216         // up to date with the leader.
217         long lastIndex = actorContext.getReplicatedLog().lastIndex();
218         leader.handleMessage(followerActor, new AppendEntriesReply(
219                 FOLLOWER_ID, newTerm, true, lastIndex, prevTerm, (short)0));
220
221         // The commit index should not get updated even though consensus was reached. This is b/c the
222         // last entry's term does match the current term. As per §5.4.1, "Raft never commits log entries
223         // from previous terms by counting replicas".
224         assertEquals("Commit Index", -1, actorContext.getCommitIndex());
225
226         followerActor.underlyingActor().clear();
227
228         // Now replicate a new entry with the new term 2.
229         long newIndex = lastIndex + 1;
230         sendReplicate(actorContext, newTerm, newIndex);
231
232         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
233         assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex());
234         assertEquals("getPrevLogTerm", prevTerm, appendEntries.getPrevLogTerm());
235         assertEquals("Entries size", 1, appendEntries.getEntries().size());
236         assertEquals("Entry getIndex", newIndex, appendEntries.getEntries().get(0).getIndex());
237         assertEquals("Entry getTerm", newTerm, appendEntries.getEntries().get(0).getTerm());
238         assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString());
239
240         // The follower replies with success. The leader should now update the commit index to the new index
241         // as per §5.4.1 "once an entry from the current term is committed by counting replicas, then all
242         // prior entries are committed indirectly".
243         leader.handleMessage(followerActor, new AppendEntriesReply(
244                 FOLLOWER_ID, newTerm, true, newIndex, newTerm, (short)0));
245
246         assertEquals("Commit Index", newIndex, actorContext.getCommitIndex());
247     }
248
249     @Test
250     public void testHandleReplicateMessageCommitIndexIncrementedBeforeConsensus() throws Exception {
251         logStart("testHandleReplicateMessageCommitIndexIncrementedBeforeConsensus");
252
253         MockRaftActorContext actorContext = createActorContextWithFollower();
254         actorContext.setRaftPolicy(createRaftPolicy(true, true));
255
256         long term = 1;
257         actorContext.getTermInformation().update(term, "");
258
259         leader = new Leader(actorContext);
260
261         // Leader will send an immediate heartbeat - ignore it.
262         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
263
264         // The follower would normally reply - simulate that explicitly here.
265         long lastIndex = actorContext.getReplicatedLog().lastIndex();
266         leader.handleMessage(followerActor, new AppendEntriesReply(
267                 FOLLOWER_ID, term, true, lastIndex, term, (short) 0));
268         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
269
270         followerActor.underlyingActor().clear();
271
272         RaftActorBehavior raftBehavior = sendReplicate(actorContext, lastIndex + 1);
273
274         // State should not change
275         assertTrue(raftBehavior instanceof Leader);
276
277         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
278         assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex());
279         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
280         assertEquals("Entries size", 1, appendEntries.getEntries().size());
281         assertEquals("Entry getIndex", lastIndex + 1, appendEntries.getEntries().get(0).getIndex());
282         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
283         assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString());
284         assertEquals("Commit Index", lastIndex+1, actorContext.getCommitIndex());
285     }
286
287     @Test
288     public void testMultipleReplicateShouldNotCauseDuplicateAppendEntriesToBeSent() throws Exception {
289         logStart("testHandleReplicateMessageSendAppendEntriesToFollower");
290
291         MockRaftActorContext actorContext = createActorContextWithFollower();
292         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
293             @Override
294             public FiniteDuration getHeartBeatInterval() {
295                 return FiniteDuration.apply(5, TimeUnit.SECONDS);
296             }
297         });
298
299         long term = 1;
300         actorContext.getTermInformation().update(term, "");
301
302         leader = new Leader(actorContext);
303
304         // Leader will send an immediate heartbeat - ignore it.
305         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
306
307         // The follower would normally reply - simulate that explicitly here.
308         long lastIndex = actorContext.getReplicatedLog().lastIndex();
309         leader.handleMessage(followerActor, new AppendEntriesReply(
310                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
311         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
312
313         followerActor.underlyingActor().clear();
314
315         for(int i=0;i<5;i++) {
316             sendReplicate(actorContext, lastIndex+i+1);
317         }
318
319         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
320         // We expect only 1 message to be sent because of two reasons,
321         // - an append entries reply was not received
322         // - the heartbeat interval has not expired
323         // In this scenario if multiple messages are sent they would likely be duplicates
324         assertEquals("The number of append entries collected should be 1", 1, allMessages.size());
325     }
326
327     @Test
328     public void testMultipleReplicateWithReplyShouldResultInAppendEntries() throws Exception {
329         logStart("testMultipleReplicateWithReplyShouldResultInAppendEntries");
330
331         MockRaftActorContext actorContext = createActorContextWithFollower();
332         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
333             @Override
334             public FiniteDuration getHeartBeatInterval() {
335                 return FiniteDuration.apply(5, TimeUnit.SECONDS);
336             }
337         });
338
339         long term = 1;
340         actorContext.getTermInformation().update(term, "");
341
342         leader = new Leader(actorContext);
343
344         // Leader will send an immediate heartbeat - ignore it.
345         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
346
347         // The follower would normally reply - simulate that explicitly here.
348         long lastIndex = actorContext.getReplicatedLog().lastIndex();
349         leader.handleMessage(followerActor, new AppendEntriesReply(
350                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
351         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
352
353         followerActor.underlyingActor().clear();
354
355         for(int i=0;i<3;i++) {
356             sendReplicate(actorContext, lastIndex+i+1);
357             leader.handleMessage(followerActor, new AppendEntriesReply(
358                     FOLLOWER_ID, term, true, lastIndex + i + 1, term, (short)0));
359
360         }
361
362         for(int i=3;i<5;i++) {
363             sendReplicate(actorContext, lastIndex + i + 1);
364         }
365
366         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
367         // We expect 4 here because the first 3 replicate got a reply and so the 4th entry would
368         // get sent to the follower - but not the 5th
369         assertEquals("The number of append entries collected should be 4", 4, allMessages.size());
370
371         for(int i=0;i<4;i++) {
372             long expected = allMessages.get(i).getEntries().get(0).getIndex();
373             assertEquals(expected, i+2);
374         }
375     }
376
377     @Test
378     public void testDuplicateAppendEntriesWillBeSentOnHeartBeat() throws Exception {
379         logStart("testDuplicateAppendEntriesWillBeSentOnHeartBeat");
380
381         MockRaftActorContext actorContext = createActorContextWithFollower();
382         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
383             @Override
384             public FiniteDuration getHeartBeatInterval() {
385                 return FiniteDuration.apply(500, TimeUnit.MILLISECONDS);
386             }
387         });
388
389         long term = 1;
390         actorContext.getTermInformation().update(term, "");
391
392         leader = new Leader(actorContext);
393
394         // Leader will send an immediate heartbeat - ignore it.
395         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
396
397         // The follower would normally reply - simulate that explicitly here.
398         long lastIndex = actorContext.getReplicatedLog().lastIndex();
399         leader.handleMessage(followerActor, new AppendEntriesReply(
400                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
401         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
402
403         followerActor.underlyingActor().clear();
404
405         sendReplicate(actorContext, lastIndex+1);
406
407         // Wait slightly longer than heartbeat duration
408         Uninterruptibles.sleepUninterruptibly(750, TimeUnit.MILLISECONDS);
409
410         leader.handleMessage(leaderActor, new SendHeartBeat());
411
412         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
413         assertEquals("The number of append entries collected should be 2", 2, allMessages.size());
414
415         assertEquals(1, allMessages.get(0).getEntries().size());
416         assertEquals(lastIndex+1, allMessages.get(0).getEntries().get(0).getIndex());
417         assertEquals(1, allMessages.get(1).getEntries().size());
418         assertEquals(lastIndex+1, allMessages.get(0).getEntries().get(0).getIndex());
419
420     }
421
422     @Test
423     public void testHeartbeatsAreAlwaysSentIfTheHeartbeatIntervalHasElapsed() throws Exception {
424         logStart("testHeartbeatsAreAlwaysSentIfTheHeartbeatIntervalHasElapsed");
425
426         MockRaftActorContext actorContext = createActorContextWithFollower();
427         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
428             @Override
429             public FiniteDuration getHeartBeatInterval() {
430                 return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
431             }
432         });
433
434         long term = 1;
435         actorContext.getTermInformation().update(term, "");
436
437         leader = new Leader(actorContext);
438
439         // Leader will send an immediate heartbeat - ignore it.
440         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
441
442         // The follower would normally reply - simulate that explicitly here.
443         long lastIndex = actorContext.getReplicatedLog().lastIndex();
444         leader.handleMessage(followerActor, new AppendEntriesReply(
445                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
446         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
447
448         followerActor.underlyingActor().clear();
449
450         for(int i=0;i<3;i++) {
451             Uninterruptibles.sleepUninterruptibly(150, TimeUnit.MILLISECONDS);
452             leader.handleMessage(leaderActor, new SendHeartBeat());
453         }
454
455         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
456         assertEquals("The number of append entries collected should be 3", 3, allMessages.size());
457     }
458
459     @Test
460     public void testSendingReplicateImmediatelyAfterHeartbeatDoesReplicate() throws Exception {
461         logStart("testSendingReplicateImmediatelyAfterHeartbeatDoesReplicate");
462
463         MockRaftActorContext actorContext = createActorContextWithFollower();
464         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
465             @Override
466             public FiniteDuration getHeartBeatInterval() {
467                 return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
468             }
469         });
470
471         long term = 1;
472         actorContext.getTermInformation().update(term, "");
473
474         leader = new Leader(actorContext);
475
476         // Leader will send an immediate heartbeat - ignore it.
477         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
478
479         // The follower would normally reply - simulate that explicitly here.
480         long lastIndex = actorContext.getReplicatedLog().lastIndex();
481         leader.handleMessage(followerActor, new AppendEntriesReply(
482                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
483         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
484
485         followerActor.underlyingActor().clear();
486
487         Uninterruptibles.sleepUninterruptibly(150, TimeUnit.MILLISECONDS);
488         leader.handleMessage(leaderActor, new SendHeartBeat());
489         sendReplicate(actorContext, lastIndex+1);
490
491         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
492         assertEquals("The number of append entries collected should be 2", 2, allMessages.size());
493
494         assertEquals(0, allMessages.get(0).getEntries().size());
495         assertEquals(1, allMessages.get(1).getEntries().size());
496     }
497
498
499     @Test
500     public void testHandleReplicateMessageWhenThereAreNoFollowers() throws Exception {
501         logStart("testHandleReplicateMessageWhenThereAreNoFollowers");
502
503         MockRaftActorContext actorContext = createActorContext();
504
505         leader = new Leader(actorContext);
506
507         actorContext.setLastApplied(0);
508
509         long newLogIndex = actorContext.getReplicatedLog().lastIndex() + 1;
510         long term = actorContext.getTermInformation().getCurrentTerm();
511         MockRaftActorContext.MockReplicatedLogEntry newEntry = new MockRaftActorContext.MockReplicatedLogEntry(
512                 term, newLogIndex, new MockRaftActorContext.MockPayload("foo"));
513
514         actorContext.getReplicatedLog().append(newEntry);
515
516         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor,
517                 new Replicate(leaderActor, "state-id", newEntry));
518
519         // State should not change
520         assertTrue(raftBehavior instanceof Leader);
521
522         assertEquals("getCommitIndex", newLogIndex, actorContext.getCommitIndex());
523
524         // We should get 2 ApplyState messages - 1 for new log entry and 1 for the previous
525         // one since lastApplied state is 0.
526         List<ApplyState> applyStateList = MessageCollectorActor.getAllMatching(
527                 leaderActor, ApplyState.class);
528         assertEquals("ApplyState count", newLogIndex, applyStateList.size());
529
530         for(int i = 0; i <= newLogIndex - 1; i++ ) {
531             ApplyState applyState = applyStateList.get(i);
532             assertEquals("getIndex", i + 1, applyState.getReplicatedLogEntry().getIndex());
533             assertEquals("getTerm", term, applyState.getReplicatedLogEntry().getTerm());
534         }
535
536         ApplyState last = applyStateList.get((int) newLogIndex - 1);
537         assertEquals("getData", newEntry.getData(), last.getReplicatedLogEntry().getData());
538         assertEquals("getIdentifier", "state-id", last.getIdentifier());
539     }
540
541     @Test
542     public void testSendAppendEntriesOnAnInProgressInstallSnapshot() throws Exception {
543         logStart("testSendAppendEntriesOnAnInProgressInstallSnapshot");
544
545         MockRaftActorContext actorContext = createActorContextWithFollower();
546
547         Map<String, String> leadersSnapshot = new HashMap<>();
548         leadersSnapshot.put("1", "A");
549         leadersSnapshot.put("2", "B");
550         leadersSnapshot.put("3", "C");
551
552         //clears leaders log
553         actorContext.getReplicatedLog().removeFrom(0);
554
555         final int commitIndex = 3;
556         final int snapshotIndex = 2;
557         final int newEntryIndex = 4;
558         final int snapshotTerm = 1;
559         final int currentTerm = 2;
560
561         // set the snapshot variables in replicatedlog
562         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
563         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
564         actorContext.setCommitIndex(commitIndex);
565         //set follower timeout to 2 mins, helps during debugging
566         actorContext.setConfigParams(new MockConfigParamsImpl(120000L, 10));
567
568         leader = new Leader(actorContext);
569
570         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
571         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
572
573         // new entry
574         ReplicatedLogImplEntry entry =
575                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
576                         new MockRaftActorContext.MockPayload("D"));
577
578         //update follower timestamp
579         leader.markFollowerActive(FOLLOWER_ID);
580
581         ByteString bs = toByteString(leadersSnapshot);
582         leader.setSnapshot(Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
583                 commitIndex, snapshotTerm, commitIndex, snapshotTerm));
584         FollowerToSnapshot fts = leader.new FollowerToSnapshot(bs);
585         leader.setFollowerSnapshot(FOLLOWER_ID, fts);
586
587         //send first chunk and no InstallSnapshotReply received yet
588         fts.getNextChunk();
589         fts.incrementChunkIndex();
590
591         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
592                 TimeUnit.MILLISECONDS);
593
594         leader.handleMessage(leaderActor, new SendHeartBeat());
595
596         AppendEntries aeproto = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
597
598         AppendEntries ae = (AppendEntries) SerializationUtils.fromSerializable(aeproto);
599
600         assertTrue("AppendEntries should be sent with empty entries", ae.getEntries().isEmpty());
601
602         //InstallSnapshotReply received
603         fts.markSendStatus(true);
604
605         leader.handleMessage(leaderActor, new SendHeartBeat());
606
607         InstallSnapshot is = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
608
609         assertEquals(commitIndex, is.getLastIncludedIndex());
610     }
611
612     @Test
613     public void testSendAppendEntriesSnapshotScenario() throws Exception {
614         logStart("testSendAppendEntriesSnapshotScenario");
615
616         MockRaftActorContext actorContext = createActorContextWithFollower();
617
618         Map<String, String> leadersSnapshot = new HashMap<>();
619         leadersSnapshot.put("1", "A");
620         leadersSnapshot.put("2", "B");
621         leadersSnapshot.put("3", "C");
622
623         //clears leaders log
624         actorContext.getReplicatedLog().removeFrom(0);
625
626         final int followersLastIndex = 2;
627         final int snapshotIndex = 3;
628         final int newEntryIndex = 4;
629         final int snapshotTerm = 1;
630         final int currentTerm = 2;
631
632         // set the snapshot variables in replicatedlog
633         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
634         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
635         actorContext.setCommitIndex(followersLastIndex);
636
637         leader = new Leader(actorContext);
638
639         // Leader will send an immediate heartbeat - ignore it.
640         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
641
642         // new entry
643         ReplicatedLogImplEntry entry =
644                 new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
645                         new MockRaftActorContext.MockPayload("D"));
646
647         actorContext.getReplicatedLog().append(entry);
648
649         //update follower timestamp
650         leader.markFollowerActive(FOLLOWER_ID);
651
652         // this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex
653         RaftActorBehavior raftBehavior = leader.handleMessage(
654                 leaderActor, new Replicate(null, "state-id", entry));
655
656         assertTrue(raftBehavior instanceof Leader);
657
658         assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
659     }
660
661     @Test
662     public void testInitiateInstallSnapshot() throws Exception {
663         logStart("testInitiateInstallSnapshot");
664
665         MockRaftActorContext actorContext = createActorContextWithFollower();
666
667         //clears leaders log
668         actorContext.getReplicatedLog().removeFrom(0);
669
670         final int followersLastIndex = 2;
671         final int snapshotIndex = 3;
672         final int newEntryIndex = 4;
673         final int snapshotTerm = 1;
674         final int currentTerm = 2;
675
676         // set the snapshot variables in replicatedlog
677         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
678         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
679         actorContext.setLastApplied(3);
680         actorContext.setCommitIndex(followersLastIndex);
681
682         leader = new Leader(actorContext);
683
684         // Leader will send an immediate heartbeat - ignore it.
685         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
686
687         // set the snapshot as absent and check if capture-snapshot is invoked.
688         leader.setSnapshot(null);
689
690         // new entry
691         ReplicatedLogImplEntry entry = new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
692                 new MockRaftActorContext.MockPayload("D"));
693
694         actorContext.getReplicatedLog().append(entry);
695
696         //update follower timestamp
697         leader.markFollowerActive(FOLLOWER_ID);
698
699         leader.handleMessage(leaderActor, new Replicate(null, "state-id", entry));
700
701         assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
702
703         CaptureSnapshot cs = actorContext.getSnapshotManager().getCaptureSnapshot();
704
705         assertTrue(cs.isInstallSnapshotInitiated());
706         assertEquals(3, cs.getLastAppliedIndex());
707         assertEquals(1, cs.getLastAppliedTerm());
708         assertEquals(4, cs.getLastIndex());
709         assertEquals(2, cs.getLastTerm());
710
711         // if an initiate is started again when first is in progress, it shouldnt initiate Capture
712         leader.handleMessage(leaderActor, new Replicate(null, "state-id", entry));
713
714         Assert.assertSame("CaptureSnapshot instance", cs, actorContext.getSnapshotManager().getCaptureSnapshot());
715     }
716
717     @Test
718     public void testInitiateForceInstallSnapshot() throws Exception {
719         logStart("testInitiateForceInstallSnapshot");
720
721         MockRaftActorContext actorContext = createActorContextWithFollower();
722
723         final int followersLastIndex = 2;
724         final int snapshotIndex = -1;
725         final int newEntryIndex = 4;
726         final int snapshotTerm = -1;
727         final int currentTerm = 2;
728
729         // set the snapshot variables in replicatedlog
730         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
731         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
732         actorContext.setLastApplied(3);
733         actorContext.setCommitIndex(followersLastIndex);
734
735         actorContext.getReplicatedLog().removeFrom(0);
736
737         leader = new Leader(actorContext);
738
739         // Leader will send an immediate heartbeat - ignore it.
740         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
741
742         // set the snapshot as absent and check if capture-snapshot is invoked.
743         leader.setSnapshot(null);
744
745         for(int i=0;i<4;i++) {
746             actorContext.getReplicatedLog().append(new ReplicatedLogImplEntry(i, 1,
747                     new MockRaftActorContext.MockPayload("X" + i)));
748         }
749
750         // new entry
751         ReplicatedLogImplEntry entry = new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
752                 new MockRaftActorContext.MockPayload("D"));
753
754         actorContext.getReplicatedLog().append(entry);
755
756         //update follower timestamp
757         leader.markFollowerActive(FOLLOWER_ID);
758
759         // Sending this AppendEntriesReply forces the Leader to capture a snapshot, which subsequently gets
760         // installed with a SendInstallSnapshot
761         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, false, 100, 1, (short) 1, true));
762
763         assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
764
765         CaptureSnapshot cs = actorContext.getSnapshotManager().getCaptureSnapshot();
766
767         assertTrue(cs.isInstallSnapshotInitiated());
768         assertEquals(3, cs.getLastAppliedIndex());
769         assertEquals(1, cs.getLastAppliedTerm());
770         assertEquals(4, cs.getLastIndex());
771         assertEquals(2, cs.getLastTerm());
772
773         // if an initiate is started again when first is in progress, it shouldnt initiate Capture
774         leader.handleMessage(leaderActor, new Replicate(null, "state-id", entry));
775
776         Assert.assertSame("CaptureSnapshot instance", cs, actorContext.getSnapshotManager().getCaptureSnapshot());
777     }
778
779
780     @Test
781     public void testInstallSnapshot() throws Exception {
782         logStart("testInstallSnapshot");
783
784         MockRaftActorContext actorContext = createActorContextWithFollower();
785
786         Map<String, String> leadersSnapshot = new HashMap<>();
787         leadersSnapshot.put("1", "A");
788         leadersSnapshot.put("2", "B");
789         leadersSnapshot.put("3", "C");
790
791         //clears leaders log
792         actorContext.getReplicatedLog().removeFrom(0);
793
794         final int lastAppliedIndex = 3;
795         final int snapshotIndex = 2;
796         final int snapshotTerm = 1;
797         final int currentTerm = 2;
798
799         // set the snapshot variables in replicatedlog
800         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
801         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
802         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
803         actorContext.setCommitIndex(lastAppliedIndex);
804         actorContext.setLastApplied(lastAppliedIndex);
805
806         leader = new Leader(actorContext);
807
808         // Initial heartbeat.
809         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
810
811         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
812         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
813
814         Snapshot snapshot = Snapshot.create(toByteString(leadersSnapshot).toByteArray(),
815                 Collections.<ReplicatedLogEntry>emptyList(),
816                 lastAppliedIndex, snapshotTerm, lastAppliedIndex, snapshotTerm);
817
818         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
819
820         assertTrue(raftBehavior instanceof Leader);
821
822         // check if installsnapshot gets called with the correct values.
823
824         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
825
826         assertNotNull(installSnapshot.getData());
827         assertEquals(lastAppliedIndex, installSnapshot.getLastIncludedIndex());
828         assertEquals(snapshotTerm, installSnapshot.getLastIncludedTerm());
829
830         assertEquals(currentTerm, installSnapshot.getTerm());
831     }
832
833     @Test
834     public void testForceInstallSnapshot() throws Exception {
835         logStart("testForceInstallSnapshot");
836
837         MockRaftActorContext actorContext = createActorContextWithFollower();
838
839         Map<String, String> leadersSnapshot = new HashMap<>();
840         leadersSnapshot.put("1", "A");
841         leadersSnapshot.put("2", "B");
842         leadersSnapshot.put("3", "C");
843
844         final int lastAppliedIndex = 3;
845         final int snapshotIndex = -1;
846         final int snapshotTerm = -1;
847         final int currentTerm = 2;
848
849         // set the snapshot variables in replicatedlog
850         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
851         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
852         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
853         actorContext.setCommitIndex(lastAppliedIndex);
854         actorContext.setLastApplied(lastAppliedIndex);
855
856         leader = new Leader(actorContext);
857
858         // Initial heartbeat.
859         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
860
861         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
862         leader.getFollower(FOLLOWER_ID).setNextIndex(-1);
863
864         Snapshot snapshot = Snapshot.create(toByteString(leadersSnapshot).toByteArray(),
865                 Collections.<ReplicatedLogEntry>emptyList(),
866                 lastAppliedIndex, snapshotTerm, lastAppliedIndex, snapshotTerm);
867
868         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
869
870         assertTrue(raftBehavior instanceof Leader);
871
872         // check if installsnapshot gets called with the correct values.
873
874         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
875
876         assertNotNull(installSnapshot.getData());
877         assertEquals(lastAppliedIndex, installSnapshot.getLastIncludedIndex());
878         assertEquals(snapshotTerm, installSnapshot.getLastIncludedTerm());
879
880         assertEquals(currentTerm, installSnapshot.getTerm());
881     }
882
883     @Test
884     public void testHandleInstallSnapshotReplyLastChunk() throws Exception {
885         logStart("testHandleInstallSnapshotReplyLastChunk");
886
887         MockRaftActorContext actorContext = createActorContextWithFollower();
888
889         final int commitIndex = 3;
890         final int snapshotIndex = 2;
891         final int snapshotTerm = 1;
892         final int currentTerm = 2;
893
894         actorContext.setCommitIndex(commitIndex);
895
896         leader = new Leader(actorContext);
897
898         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
899         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
900
901         // Ignore initial heartbeat.
902         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
903
904         Map<String, String> leadersSnapshot = new HashMap<>();
905         leadersSnapshot.put("1", "A");
906         leadersSnapshot.put("2", "B");
907         leadersSnapshot.put("3", "C");
908
909         // set the snapshot variables in replicatedlog
910
911         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
912         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
913         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
914
915         ByteString bs = toByteString(leadersSnapshot);
916         leader.setSnapshot(Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
917                 commitIndex, snapshotTerm, commitIndex, snapshotTerm));
918         FollowerToSnapshot fts = leader.new FollowerToSnapshot(bs);
919         leader.setFollowerSnapshot(FOLLOWER_ID, fts);
920         while(!fts.isLastChunk(fts.getChunkIndex())) {
921             fts.getNextChunk();
922             fts.incrementChunkIndex();
923         }
924
925         //clears leaders log
926         actorContext.getReplicatedLog().removeFrom(0);
927
928         RaftActorBehavior raftBehavior = leader.handleMessage(followerActor,
929                 new InstallSnapshotReply(currentTerm, FOLLOWER_ID, fts.getChunkIndex(), true));
930
931         assertTrue(raftBehavior instanceof Leader);
932
933         assertEquals(0, leader.followerSnapshotSize());
934         assertEquals(1, leader.followerLogSize());
935         FollowerLogInformation fli = leader.getFollower(FOLLOWER_ID);
936         assertNotNull(fli);
937         assertEquals(commitIndex, fli.getMatchIndex());
938         assertEquals(commitIndex + 1, fli.getNextIndex());
939     }
940
941     @Test
942     public void testSendSnapshotfromInstallSnapshotReply() throws Exception {
943         logStart("testSendSnapshotfromInstallSnapshotReply");
944
945         MockRaftActorContext actorContext = createActorContextWithFollower();
946
947         final int commitIndex = 3;
948         final int snapshotIndex = 2;
949         final int snapshotTerm = 1;
950         final int currentTerm = 2;
951
952         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(){
953             @Override
954             public int getSnapshotChunkSize() {
955                 return 50;
956             }
957         };
958         configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS));
959         configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
960
961         actorContext.setConfigParams(configParams);
962         actorContext.setCommitIndex(commitIndex);
963
964         leader = new Leader(actorContext);
965
966         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
967         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
968
969         Map<String, String> leadersSnapshot = new HashMap<>();
970         leadersSnapshot.put("1", "A");
971         leadersSnapshot.put("2", "B");
972         leadersSnapshot.put("3", "C");
973
974         // set the snapshot variables in replicatedlog
975         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
976         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
977         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
978
979         ByteString bs = toByteString(leadersSnapshot);
980         Snapshot snapshot = Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
981                 commitIndex, snapshotTerm, commitIndex, snapshotTerm);
982         leader.setSnapshot(snapshot);
983
984         leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
985
986         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
987
988         assertEquals(1, installSnapshot.getChunkIndex());
989         assertEquals(3, installSnapshot.getTotalChunks());
990
991         followerActor.underlyingActor().clear();
992         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
993                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
994
995         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
996
997         assertEquals(2, installSnapshot.getChunkIndex());
998         assertEquals(3, installSnapshot.getTotalChunks());
999
1000         followerActor.underlyingActor().clear();
1001         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
1002                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
1003
1004         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1005
1006         // Send snapshot reply one more time and make sure that a new snapshot message should not be sent to follower
1007         followerActor.underlyingActor().clear();
1008         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
1009                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
1010
1011         installSnapshot = MessageCollectorActor.getFirstMatching(followerActor, InstallSnapshot.class);
1012
1013         Assert.assertNull(installSnapshot);
1014     }
1015
1016
1017     @Test
1018     public void testHandleInstallSnapshotReplyWithInvalidChunkIndex() throws Exception{
1019         logStart("testHandleInstallSnapshotReplyWithInvalidChunkIndex");
1020
1021         MockRaftActorContext actorContext = createActorContextWithFollower();
1022
1023         final int commitIndex = 3;
1024         final int snapshotIndex = 2;
1025         final int snapshotTerm = 1;
1026         final int currentTerm = 2;
1027
1028         actorContext.setConfigParams(new DefaultConfigParamsImpl(){
1029             @Override
1030             public int getSnapshotChunkSize() {
1031                 return 50;
1032             }
1033         });
1034
1035         actorContext.setCommitIndex(commitIndex);
1036
1037         leader = new Leader(actorContext);
1038
1039         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
1040         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
1041
1042         Map<String, String> leadersSnapshot = new HashMap<>();
1043         leadersSnapshot.put("1", "A");
1044         leadersSnapshot.put("2", "B");
1045         leadersSnapshot.put("3", "C");
1046
1047         // set the snapshot variables in replicatedlog
1048         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
1049         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
1050         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
1051
1052         ByteString bs = toByteString(leadersSnapshot);
1053         Snapshot snapshot = Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
1054                 commitIndex, snapshotTerm, commitIndex, snapshotTerm);
1055         leader.setSnapshot(snapshot);
1056
1057         Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
1058         leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
1059
1060         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1061
1062         assertEquals(1, installSnapshot.getChunkIndex());
1063         assertEquals(3, installSnapshot.getTotalChunks());
1064
1065         followerActor.underlyingActor().clear();
1066
1067         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
1068                 FOLLOWER_ID, -1, false));
1069
1070         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
1071                 TimeUnit.MILLISECONDS);
1072
1073         leader.handleMessage(leaderActor, new SendHeartBeat());
1074
1075         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1076
1077         assertEquals(1, installSnapshot.getChunkIndex());
1078         assertEquals(3, installSnapshot.getTotalChunks());
1079     }
1080
1081     @Test
1082     public void testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk() throws Exception {
1083         logStart("testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk");
1084
1085         MockRaftActorContext actorContext = createActorContextWithFollower();
1086
1087         final int commitIndex = 3;
1088         final int snapshotIndex = 2;
1089         final int snapshotTerm = 1;
1090         final int currentTerm = 2;
1091
1092         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
1093             @Override
1094             public int getSnapshotChunkSize() {
1095                 return 50;
1096             }
1097         });
1098
1099         actorContext.setCommitIndex(commitIndex);
1100
1101         leader = new Leader(actorContext);
1102
1103         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
1104         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
1105
1106         Map<String, String> leadersSnapshot = new HashMap<>();
1107         leadersSnapshot.put("1", "A");
1108         leadersSnapshot.put("2", "B");
1109         leadersSnapshot.put("3", "C");
1110
1111         // set the snapshot variables in replicatedlog
1112         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
1113         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
1114         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
1115
1116         ByteString bs = toByteString(leadersSnapshot);
1117         Snapshot snapshot = Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
1118                 commitIndex, snapshotTerm, commitIndex, snapshotTerm);
1119         leader.setSnapshot(snapshot);
1120
1121         leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
1122
1123         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1124
1125         assertEquals(1, installSnapshot.getChunkIndex());
1126         assertEquals(3, installSnapshot.getTotalChunks());
1127         assertEquals(AbstractLeader.INITIAL_LAST_CHUNK_HASH_CODE, installSnapshot.getLastChunkHashCode().get().intValue());
1128
1129         int hashCode = Arrays.hashCode(installSnapshot.getData());
1130
1131         followerActor.underlyingActor().clear();
1132
1133         leader.handleMessage(followerActor, new InstallSnapshotReply(installSnapshot.getTerm(),
1134                 FOLLOWER_ID, 1, true));
1135
1136         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1137
1138         assertEquals(2, installSnapshot.getChunkIndex());
1139         assertEquals(3, installSnapshot.getTotalChunks());
1140         assertEquals(hashCode, installSnapshot.getLastChunkHashCode().get().intValue());
1141     }
1142
1143     @Test
1144     public void testFollowerToSnapshotLogic() {
1145         logStart("testFollowerToSnapshotLogic");
1146
1147         MockRaftActorContext actorContext = createActorContext();
1148
1149         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
1150             @Override
1151             public int getSnapshotChunkSize() {
1152                 return 50;
1153             }
1154         });
1155
1156         leader = new Leader(actorContext);
1157
1158         Map<String, String> leadersSnapshot = new HashMap<>();
1159         leadersSnapshot.put("1", "A");
1160         leadersSnapshot.put("2", "B");
1161         leadersSnapshot.put("3", "C");
1162
1163         ByteString bs = toByteString(leadersSnapshot);
1164         byte[] barray = bs.toByteArray();
1165
1166         FollowerToSnapshot fts = leader.new FollowerToSnapshot(bs);
1167         leader.setFollowerSnapshot(FOLLOWER_ID, fts);
1168
1169         assertEquals(bs.size(), barray.length);
1170
1171         int chunkIndex=0;
1172         for (int i=0; i < barray.length; i = i + 50) {
1173             int j = i + 50;
1174             chunkIndex++;
1175
1176             if (i + 50 > barray.length) {
1177                 j = barray.length;
1178             }
1179
1180             byte[] chunk = fts.getNextChunk();
1181             assertEquals("bytestring size not matching for chunk:"+ chunkIndex, j-i, chunk.length);
1182             assertEquals("chunkindex not matching", chunkIndex, fts.getChunkIndex());
1183
1184             fts.markSendStatus(true);
1185             if (!fts.isLastChunk(chunkIndex)) {
1186                 fts.incrementChunkIndex();
1187             }
1188         }
1189
1190         assertEquals("totalChunks not matching", chunkIndex, fts.getTotalChunks());
1191     }
1192
1193     @Override protected RaftActorBehavior createBehavior(
1194         RaftActorContext actorContext) {
1195         return new Leader(actorContext);
1196     }
1197
1198     @Override
1199     protected MockRaftActorContext createActorContext() {
1200         return createActorContext(leaderActor);
1201     }
1202
1203     @Override
1204     protected MockRaftActorContext createActorContext(ActorRef actorRef) {
1205         return createActorContext(LEADER_ID, actorRef);
1206     }
1207
1208     private MockRaftActorContext createActorContextWithFollower() {
1209         MockRaftActorContext actorContext = createActorContext();
1210         actorContext.setPeerAddresses(ImmutableMap.<String, String>builder().put(FOLLOWER_ID,
1211                 followerActor.path().toString()).build());
1212         return actorContext;
1213     }
1214
1215     private MockRaftActorContext createActorContext(String id, ActorRef actorRef) {
1216         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
1217         configParams.setHeartBeatInterval(new FiniteDuration(50, TimeUnit.MILLISECONDS));
1218         configParams.setElectionTimeoutFactor(100000);
1219         MockRaftActorContext context = new MockRaftActorContext(id, getSystem(), actorRef);
1220         context.setConfigParams(configParams);
1221         context.setPayloadVersion(payloadVersion);
1222         return context;
1223     }
1224
1225     private MockRaftActorContext createFollowerActorContextWithLeader() {
1226         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
1227         DefaultConfigParamsImpl followerConfig = new DefaultConfigParamsImpl();
1228         followerConfig.setElectionTimeoutFactor(10000);
1229         followerActorContext.setConfigParams(followerConfig);
1230         followerActorContext.setPeerAddresses(ImmutableMap.of(LEADER_ID, leaderActor.path().toString()));
1231         return followerActorContext;
1232     }
1233
1234     @Test
1235     public void testLeaderCreatedWithCommitIndexLessThanLastIndex() throws Exception {
1236         logStart("testLeaderCreatedWithCommitIndexLessThanLastIndex");
1237
1238         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1239
1240         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
1241
1242         Follower follower = new Follower(followerActorContext);
1243         followerActor.underlyingActor().setBehavior(follower);
1244
1245         Map<String, String> peerAddresses = new HashMap<>();
1246         peerAddresses.put(FOLLOWER_ID, followerActor.path().toString());
1247
1248         leaderActorContext.setPeerAddresses(peerAddresses);
1249
1250         leaderActorContext.getReplicatedLog().removeFrom(0);
1251
1252         //create 3 entries
1253         leaderActorContext.setReplicatedLog(
1254                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1255
1256         leaderActorContext.setCommitIndex(1);
1257
1258         followerActorContext.getReplicatedLog().removeFrom(0);
1259
1260         // follower too has the exact same log entries and has the same commit index
1261         followerActorContext.setReplicatedLog(
1262                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1263
1264         followerActorContext.setCommitIndex(1);
1265
1266         leader = new Leader(leaderActorContext);
1267
1268         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1269
1270         assertEquals(1, appendEntries.getLeaderCommit());
1271         assertEquals(0, appendEntries.getEntries().size());
1272         assertEquals(0, appendEntries.getPrevLogIndex());
1273
1274         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(
1275                 leaderActor, AppendEntriesReply.class);
1276
1277         assertEquals(2, appendEntriesReply.getLogLastIndex());
1278         assertEquals(1, appendEntriesReply.getLogLastTerm());
1279
1280         // follower returns its next index
1281         assertEquals(2, appendEntriesReply.getLogLastIndex());
1282         assertEquals(1, appendEntriesReply.getLogLastTerm());
1283
1284         follower.close();
1285     }
1286
1287     @Test
1288     public void testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex() throws Exception {
1289         logStart("testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex");
1290
1291         MockRaftActorContext leaderActorContext = createActorContext();
1292
1293         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
1294         followerActorContext.setPeerAddresses(ImmutableMap.of(LEADER_ID, leaderActor.path().toString()));
1295
1296         Follower follower = new Follower(followerActorContext);
1297         followerActor.underlyingActor().setBehavior(follower);
1298
1299         Map<String, String> leaderPeerAddresses = new HashMap<>();
1300         leaderPeerAddresses.put(FOLLOWER_ID, followerActor.path().toString());
1301
1302         leaderActorContext.setPeerAddresses(leaderPeerAddresses);
1303
1304         leaderActorContext.getReplicatedLog().removeFrom(0);
1305
1306         leaderActorContext.setReplicatedLog(
1307                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1308
1309         leaderActorContext.setCommitIndex(1);
1310
1311         followerActorContext.getReplicatedLog().removeFrom(0);
1312
1313         followerActorContext.setReplicatedLog(
1314                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1315
1316         // follower has the same log entries but its commit index > leaders commit index
1317         followerActorContext.setCommitIndex(2);
1318
1319         leader = new Leader(leaderActorContext);
1320
1321         // Initial heartbeat
1322         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1323
1324         assertEquals(1, appendEntries.getLeaderCommit());
1325         assertEquals(0, appendEntries.getEntries().size());
1326         assertEquals(0, appendEntries.getPrevLogIndex());
1327
1328         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(
1329                 leaderActor, AppendEntriesReply.class);
1330
1331         assertEquals(2, appendEntriesReply.getLogLastIndex());
1332         assertEquals(1, appendEntriesReply.getLogLastTerm());
1333
1334         leaderActor.underlyingActor().setBehavior(follower);
1335         leader.handleMessage(followerActor, appendEntriesReply);
1336
1337         leaderActor.underlyingActor().clear();
1338         followerActor.underlyingActor().clear();
1339
1340         Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(),
1341                 TimeUnit.MILLISECONDS);
1342
1343         leader.handleMessage(leaderActor, new SendHeartBeat());
1344
1345         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1346
1347         assertEquals(2, appendEntries.getLeaderCommit());
1348         assertEquals(0, appendEntries.getEntries().size());
1349         assertEquals(2, appendEntries.getPrevLogIndex());
1350
1351         appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1352
1353         assertEquals(2, appendEntriesReply.getLogLastIndex());
1354         assertEquals(1, appendEntriesReply.getLogLastTerm());
1355
1356         assertEquals(2, followerActorContext.getCommitIndex());
1357
1358         follower.close();
1359     }
1360
1361     @Test
1362     public void testHandleAppendEntriesReplyFailureWithFollowersLogBehindTheLeader(){
1363         logStart("testHandleAppendEntriesReplyFailureWithFollowersLogBehindTheLeader");
1364
1365         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1366         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1367                 new FiniteDuration(1000, TimeUnit.SECONDS));
1368
1369         leaderActorContext.setReplicatedLog(
1370                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1371         long leaderCommitIndex = 2;
1372         leaderActorContext.setCommitIndex(leaderCommitIndex);
1373         leaderActorContext.setLastApplied(leaderCommitIndex);
1374
1375         ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1376         ReplicatedLogEntry leadersThirdLogEntry = leaderActorContext.getReplicatedLog().get(2);
1377
1378         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1379
1380         followerActorContext.setReplicatedLog(
1381                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build());
1382         followerActorContext.setCommitIndex(0);
1383         followerActorContext.setLastApplied(0);
1384
1385         Follower follower = new Follower(followerActorContext);
1386         followerActor.underlyingActor().setBehavior(follower);
1387
1388         leader = new Leader(leaderActorContext);
1389
1390         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1391         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1392
1393         MessageCollectorActor.clearMessages(followerActor);
1394         MessageCollectorActor.clearMessages(leaderActor);
1395
1396         // Verify initial AppendEntries sent with the leader's current commit index.
1397         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1398         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1399         assertEquals("getPrevLogIndex", 1, appendEntries.getPrevLogIndex());
1400
1401         leaderActor.underlyingActor().setBehavior(leader);
1402
1403         leader.handleMessage(followerActor, appendEntriesReply);
1404
1405         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1);
1406         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1407
1408         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1409         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1410         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1411
1412         assertEquals("First entry index", 1, appendEntries.getEntries().get(0).getIndex());
1413         assertEquals("First entry data", leadersSecondLogEntry.getData(),
1414                 appendEntries.getEntries().get(0).getData());
1415         assertEquals("Second entry index", 2, appendEntries.getEntries().get(1).getIndex());
1416         assertEquals("Second entry data", leadersThirdLogEntry.getData(),
1417                 appendEntries.getEntries().get(1).getData());
1418
1419         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1420         assertEquals("getNextIndex", 3, followerInfo.getNextIndex());
1421
1422         List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2);
1423
1424         ApplyState applyState = applyStateList.get(0);
1425         assertEquals("Follower's first ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex());
1426         assertEquals("Follower's first ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1427         assertEquals("Follower's first ApplyState data", leadersSecondLogEntry.getData(),
1428                 applyState.getReplicatedLogEntry().getData());
1429
1430         applyState = applyStateList.get(1);
1431         assertEquals("Follower's second ApplyState index", 2, applyState.getReplicatedLogEntry().getIndex());
1432         assertEquals("Follower's second ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1433         assertEquals("Follower's second ApplyState data", leadersThirdLogEntry.getData(),
1434                 applyState.getReplicatedLogEntry().getData());
1435
1436         assertEquals("Follower's commit index", 2, followerActorContext.getCommitIndex());
1437         assertEquals("Follower's lastIndex", 2, followerActorContext.getReplicatedLog().lastIndex());
1438     }
1439
1440     @Test
1441     public void testHandleAppendEntriesReplyFailureWithFollowersLogEmpty() {
1442         logStart("testHandleAppendEntriesReplyFailureWithFollowersLogEmpty");
1443
1444         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1445         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1446                 new FiniteDuration(1000, TimeUnit.SECONDS));
1447
1448         leaderActorContext.setReplicatedLog(
1449                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build());
1450         long leaderCommitIndex = 1;
1451         leaderActorContext.setCommitIndex(leaderCommitIndex);
1452         leaderActorContext.setLastApplied(leaderCommitIndex);
1453
1454         ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0);
1455         ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1456
1457         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1458
1459         followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
1460         followerActorContext.setCommitIndex(-1);
1461         followerActorContext.setLastApplied(-1);
1462
1463         Follower follower = new Follower(followerActorContext);
1464         followerActor.underlyingActor().setBehavior(follower);
1465
1466         leader = new Leader(leaderActorContext);
1467
1468         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1469         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1470
1471         MessageCollectorActor.clearMessages(followerActor);
1472         MessageCollectorActor.clearMessages(leaderActor);
1473
1474         // Verify initial AppendEntries sent with the leader's current commit index.
1475         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1476         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1477         assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex());
1478
1479         leaderActor.underlyingActor().setBehavior(leader);
1480
1481         leader.handleMessage(followerActor, appendEntriesReply);
1482
1483         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1);
1484         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1485
1486         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1487         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1488         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1489
1490         assertEquals("First entry index", 0, appendEntries.getEntries().get(0).getIndex());
1491         assertEquals("First entry data", leadersFirstLogEntry.getData(),
1492                 appendEntries.getEntries().get(0).getData());
1493         assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex());
1494         assertEquals("Second entry data", leadersSecondLogEntry.getData(),
1495                 appendEntries.getEntries().get(1).getData());
1496
1497         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1498         assertEquals("getNextIndex", 2, followerInfo.getNextIndex());
1499
1500         List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2);
1501
1502         ApplyState applyState = applyStateList.get(0);
1503         assertEquals("Follower's first ApplyState index", 0, applyState.getReplicatedLogEntry().getIndex());
1504         assertEquals("Follower's first ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1505         assertEquals("Follower's first ApplyState data", leadersFirstLogEntry.getData(),
1506                 applyState.getReplicatedLogEntry().getData());
1507
1508         applyState = applyStateList.get(1);
1509         assertEquals("Follower's second ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex());
1510         assertEquals("Follower's second ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1511         assertEquals("Follower's second ApplyState data", leadersSecondLogEntry.getData(),
1512                 applyState.getReplicatedLogEntry().getData());
1513
1514         assertEquals("Follower's commit index", 1, followerActorContext.getCommitIndex());
1515         assertEquals("Follower's lastIndex", 1, followerActorContext.getReplicatedLog().lastIndex());
1516     }
1517
1518     @Test
1519     public void testHandleAppendEntriesReplyFailureWithFollowersLogTermDifferent(){
1520         logStart("testHandleAppendEntriesReplyFailureWithFollowersLogTermDifferent");
1521
1522         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1523         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1524                 new FiniteDuration(1000, TimeUnit.SECONDS));
1525
1526         leaderActorContext.setReplicatedLog(
1527                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 2).build());
1528         long leaderCommitIndex = 1;
1529         leaderActorContext.setCommitIndex(leaderCommitIndex);
1530         leaderActorContext.setLastApplied(leaderCommitIndex);
1531
1532         ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0);
1533         ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1534
1535         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1536
1537         followerActorContext.setReplicatedLog(
1538                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build());
1539         followerActorContext.setCommitIndex(-1);
1540         followerActorContext.setLastApplied(-1);
1541
1542         Follower follower = new Follower(followerActorContext);
1543         followerActor.underlyingActor().setBehavior(follower);
1544
1545         leader = new Leader(leaderActorContext);
1546
1547         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1548         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1549
1550         MessageCollectorActor.clearMessages(followerActor);
1551         MessageCollectorActor.clearMessages(leaderActor);
1552
1553         // Verify initial AppendEntries sent with the leader's current commit index.
1554         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1555         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1556         assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex());
1557
1558         leaderActor.underlyingActor().setBehavior(leader);
1559
1560         leader.handleMessage(followerActor, appendEntriesReply);
1561
1562         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1);
1563         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1564
1565         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1566         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1567         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1568
1569         assertEquals("First entry index", 0, appendEntries.getEntries().get(0).getIndex());
1570         assertEquals("First entry term", 2, appendEntries.getEntries().get(0).getTerm());
1571         assertEquals("First entry data", leadersFirstLogEntry.getData(),
1572                 appendEntries.getEntries().get(0).getData());
1573         assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex());
1574         assertEquals("Second entry term", 2, appendEntries.getEntries().get(1).getTerm());
1575         assertEquals("Second entry data", leadersSecondLogEntry.getData(),
1576                 appendEntries.getEntries().get(1).getData());
1577
1578         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1579         assertEquals("getNextIndex", 2, followerInfo.getNextIndex());
1580
1581         List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2);
1582
1583         ApplyState applyState = applyStateList.get(0);
1584         assertEquals("Follower's first ApplyState index", 0, applyState.getReplicatedLogEntry().getIndex());
1585         assertEquals("Follower's first ApplyState term", 2, applyState.getReplicatedLogEntry().getTerm());
1586         assertEquals("Follower's first ApplyState data", leadersFirstLogEntry.getData(),
1587                 applyState.getReplicatedLogEntry().getData());
1588
1589         applyState = applyStateList.get(1);
1590         assertEquals("Follower's second ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex());
1591         assertEquals("Follower's second ApplyState term", 2, applyState.getReplicatedLogEntry().getTerm());
1592         assertEquals("Follower's second ApplyState data", leadersSecondLogEntry.getData(),
1593                 applyState.getReplicatedLogEntry().getData());
1594
1595         assertEquals("Follower's commit index", 1, followerActorContext.getCommitIndex());
1596         assertEquals("Follower's lastIndex", 1, followerActorContext.getReplicatedLog().lastIndex());
1597         assertEquals("Follower's lastTerm", 2, followerActorContext.getReplicatedLog().lastTerm());
1598     }
1599
1600     @Test
1601     public void testHandleAppendEntriesReplyWithNewerTerm(){
1602         logStart("testHandleAppendEntriesReplyWithNewerTerm");
1603
1604         MockRaftActorContext leaderActorContext = createActorContext();
1605         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1606                 new FiniteDuration(10000, TimeUnit.SECONDS));
1607
1608         leaderActorContext.setReplicatedLog(
1609                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 2).build());
1610
1611         leader = new Leader(leaderActorContext);
1612         leaderActor.underlyingActor().setBehavior(leader);
1613         leaderActor.tell(new AppendEntriesReply("foo", 20, false, 1000, 10, (short) 1), ActorRef.noSender());
1614
1615         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1616
1617         assertEquals(false, appendEntriesReply.isSuccess());
1618         assertEquals(RaftState.Follower, leaderActor.underlyingActor().getFirstBehaviorChange().state());
1619
1620         MessageCollectorActor.clearMessages(leaderActor);
1621     }
1622
1623     @Test
1624     public void testHandleAppendEntriesReplyWithNewerTermWhenElectionsAreDisabled(){
1625         logStart("testHandleAppendEntriesReplyWithNewerTermWhenElectionsAreDisabled");
1626
1627         MockRaftActorContext leaderActorContext = createActorContext();
1628         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1629                 new FiniteDuration(10000, TimeUnit.SECONDS));
1630
1631         leaderActorContext.setReplicatedLog(
1632                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 2).build());
1633         leaderActorContext.setRaftPolicy(createRaftPolicy(false, false));
1634
1635         leader = new Leader(leaderActorContext);
1636         leaderActor.underlyingActor().setBehavior(leader);
1637         leaderActor.tell(new AppendEntriesReply("foo", 20, false, 1000, 10, (short) 1), ActorRef.noSender());
1638
1639         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1640
1641         assertEquals(false, appendEntriesReply.isSuccess());
1642         assertEquals(RaftState.Leader, leaderActor.underlyingActor().getFirstBehaviorChange().state());
1643
1644         MessageCollectorActor.clearMessages(leaderActor);
1645     }
1646
1647     @Test
1648     public void testHandleAppendEntriesReplySuccess() throws Exception {
1649         logStart("testHandleAppendEntriesReplySuccess");
1650
1651         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1652
1653         leaderActorContext.setReplicatedLog(
1654                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1655
1656         leaderActorContext.setCommitIndex(1);
1657         leaderActorContext.setLastApplied(1);
1658         leaderActorContext.getTermInformation().update(1, "leader");
1659
1660         leader = new Leader(leaderActorContext);
1661
1662         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1663
1664         assertEquals(payloadVersion, leader.getLeaderPayloadVersion());
1665         assertEquals(RaftVersions.HELIUM_VERSION, followerInfo.getRaftVersion());
1666
1667         short payloadVersion = 5;
1668         AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, true, 2, 1, payloadVersion);
1669
1670         RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
1671
1672         assertEquals(RaftState.Leader, raftActorBehavior.state());
1673
1674         assertEquals(2, leaderActorContext.getCommitIndex());
1675
1676         ApplyJournalEntries applyJournalEntries = MessageCollectorActor.expectFirstMatching(
1677                 leaderActor, ApplyJournalEntries.class);
1678
1679         assertEquals(2, leaderActorContext.getLastApplied());
1680
1681         assertEquals(2, applyJournalEntries.getToIndex());
1682
1683         List<ApplyState> applyStateList = MessageCollectorActor.getAllMatching(leaderActor,
1684                 ApplyState.class);
1685
1686         assertEquals(1,applyStateList.size());
1687
1688         ApplyState applyState = applyStateList.get(0);
1689
1690         assertEquals(2, applyState.getReplicatedLogEntry().getIndex());
1691
1692         assertEquals(2, followerInfo.getMatchIndex());
1693         assertEquals(3, followerInfo.getNextIndex());
1694         assertEquals(payloadVersion, followerInfo.getPayloadVersion());
1695         assertEquals(RaftVersions.CURRENT_VERSION, followerInfo.getRaftVersion());
1696     }
1697
1698     @Test
1699     public void testHandleAppendEntriesReplyUnknownFollower(){
1700         logStart("testHandleAppendEntriesReplyUnknownFollower");
1701
1702         MockRaftActorContext leaderActorContext = createActorContext();
1703
1704         leader = new Leader(leaderActorContext);
1705
1706         AppendEntriesReply reply = new AppendEntriesReply("unkown-follower", 1, false, 10, 1, (short)0);
1707
1708         RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
1709
1710         assertEquals(RaftState.Leader, raftActorBehavior.state());
1711     }
1712
1713     @Test
1714     public void testFollowerCatchUpWithAppendEntriesMaxDataSizeExceeded() {
1715         logStart("testFollowerCatchUpWithAppendEntriesMaxDataSizeExceeded");
1716
1717         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1718         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1719                 new FiniteDuration(1000, TimeUnit.SECONDS));
1720         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setSnapshotChunkSize(2);
1721
1722         leaderActorContext.setReplicatedLog(
1723                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 4, 1).build());
1724         long leaderCommitIndex = 3;
1725         leaderActorContext.setCommitIndex(leaderCommitIndex);
1726         leaderActorContext.setLastApplied(leaderCommitIndex);
1727
1728         ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0);
1729         ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1730         ReplicatedLogEntry leadersThirdLogEntry = leaderActorContext.getReplicatedLog().get(2);
1731         ReplicatedLogEntry leadersFourthLogEntry = leaderActorContext.getReplicatedLog().get(3);
1732
1733         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1734
1735         followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
1736         followerActorContext.setCommitIndex(-1);
1737         followerActorContext.setLastApplied(-1);
1738
1739         Follower follower = new Follower(followerActorContext);
1740         followerActor.underlyingActor().setBehavior(follower);
1741
1742         leader = new Leader(leaderActorContext);
1743
1744         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1745         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1746
1747         MessageCollectorActor.clearMessages(followerActor);
1748         MessageCollectorActor.clearMessages(leaderActor);
1749
1750         // Verify initial AppendEntries sent with the leader's current commit index.
1751         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1752         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1753         assertEquals("getPrevLogIndex", 2, appendEntries.getPrevLogIndex());
1754
1755         leaderActor.underlyingActor().setBehavior(leader);
1756
1757         leader.handleMessage(followerActor, appendEntriesReply);
1758
1759         List<AppendEntries> appendEntriesList = MessageCollectorActor.expectMatching(followerActor, AppendEntries.class, 2);
1760         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 2);
1761
1762         appendEntries = appendEntriesList.get(0);
1763         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1764         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1765         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1766
1767         assertEquals("First entry index", 0, appendEntries.getEntries().get(0).getIndex());
1768         assertEquals("First entry data", leadersFirstLogEntry.getData(),
1769                 appendEntries.getEntries().get(0).getData());
1770         assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex());
1771         assertEquals("Second entry data", leadersSecondLogEntry.getData(),
1772                 appendEntries.getEntries().get(1).getData());
1773
1774         appendEntries = appendEntriesList.get(1);
1775         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1776         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1777         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1778
1779         assertEquals("First entry index", 2, appendEntries.getEntries().get(0).getIndex());
1780         assertEquals("First entry data", leadersThirdLogEntry.getData(),
1781                 appendEntries.getEntries().get(0).getData());
1782         assertEquals("Second entry index", 3, appendEntries.getEntries().get(1).getIndex());
1783         assertEquals("Second entry data", leadersFourthLogEntry.getData(),
1784                 appendEntries.getEntries().get(1).getData());
1785
1786         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1787         assertEquals("getNextIndex", 4, followerInfo.getNextIndex());
1788
1789         MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 4);
1790
1791         assertEquals("Follower's commit index", 3, followerActorContext.getCommitIndex());
1792         assertEquals("Follower's lastIndex", 3, followerActorContext.getReplicatedLog().lastIndex());
1793     }
1794
1795     @Test
1796     public void testHandleRequestVoteReply(){
1797         logStart("testHandleRequestVoteReply");
1798
1799         MockRaftActorContext leaderActorContext = createActorContext();
1800
1801         leader = new Leader(leaderActorContext);
1802
1803         // Should be a no-op.
1804         RaftActorBehavior raftActorBehavior = leader.handleRequestVoteReply(followerActor,
1805                 new RequestVoteReply(1, true));
1806
1807         assertEquals(RaftState.Leader, raftActorBehavior.state());
1808
1809         raftActorBehavior = leader.handleRequestVoteReply(followerActor, new RequestVoteReply(1, false));
1810
1811         assertEquals(RaftState.Leader, raftActorBehavior.state());
1812     }
1813
1814     @Test
1815     public void testIsolatedLeaderCheckNoFollowers() {
1816         logStart("testIsolatedLeaderCheckNoFollowers");
1817
1818         MockRaftActorContext leaderActorContext = createActorContext();
1819
1820         leader = new Leader(leaderActorContext);
1821         RaftActorBehavior behavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1822         Assert.assertTrue(behavior instanceof Leader);
1823     }
1824
1825     private RaftActorBehavior setupIsolatedLeaderCheckTestWithTwoFollowers(RaftPolicy raftPolicy){
1826         ActorRef followerActor1 = getSystem().actorOf(MessageCollectorActor.props(), "follower-1");
1827         ActorRef followerActor2 = getSystem().actorOf(MessageCollectorActor.props(), "follower-2");
1828
1829         MockRaftActorContext leaderActorContext = createActorContext();
1830
1831         Map<String, String> peerAddresses = new HashMap<>();
1832         peerAddresses.put("follower-1", followerActor1.path().toString());
1833         peerAddresses.put("follower-2", followerActor2.path().toString());
1834
1835         leaderActorContext.setPeerAddresses(peerAddresses);
1836         leaderActorContext.setRaftPolicy(raftPolicy);
1837
1838         leader = new Leader(leaderActorContext);
1839
1840         leader.markFollowerActive("follower-1");
1841         leader.markFollowerActive("follower-2");
1842         RaftActorBehavior behavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1843         Assert.assertTrue("Behavior not instance of Leader when all followers are active",
1844                 behavior instanceof Leader);
1845
1846         // kill 1 follower and verify if that got killed
1847         final JavaTestKit probe = new JavaTestKit(getSystem());
1848         probe.watch(followerActor1);
1849         followerActor1.tell(PoisonPill.getInstance(), ActorRef.noSender());
1850         final Terminated termMsg1 = probe.expectMsgClass(Terminated.class);
1851         assertEquals(termMsg1.getActor(), followerActor1);
1852
1853         leader.markFollowerInActive("follower-1");
1854         leader.markFollowerActive("follower-2");
1855         behavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1856         Assert.assertTrue("Behavior not instance of Leader when majority of followers are active",
1857                 behavior instanceof Leader);
1858
1859         // kill 2nd follower and leader should change to Isolated leader
1860         followerActor2.tell(PoisonPill.getInstance(), null);
1861         probe.watch(followerActor2);
1862         followerActor2.tell(PoisonPill.getInstance(), ActorRef.noSender());
1863         final Terminated termMsg2 = probe.expectMsgClass(Terminated.class);
1864         assertEquals(termMsg2.getActor(), followerActor2);
1865
1866         leader.markFollowerInActive("follower-2");
1867         return leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1868     }
1869
1870     @Test
1871     public void testIsolatedLeaderCheckTwoFollowers() throws Exception {
1872         logStart("testIsolatedLeaderCheckTwoFollowers");
1873
1874         RaftActorBehavior behavior = setupIsolatedLeaderCheckTestWithTwoFollowers(DefaultRaftPolicy.INSTANCE);
1875
1876         Assert.assertTrue("Behavior not instance of IsolatedLeader when majority followers are inactive",
1877             behavior instanceof IsolatedLeader);
1878     }
1879
1880     @Test
1881     public void testIsolatedLeaderCheckTwoFollowersWhenElectionsAreDisabled() throws Exception {
1882         logStart("testIsolatedLeaderCheckTwoFollowersWhenElectionsAreDisabled");
1883
1884         RaftActorBehavior behavior = setupIsolatedLeaderCheckTestWithTwoFollowers(createRaftPolicy(false, true));
1885
1886         Assert.assertTrue("Behavior should not switch to IsolatedLeader because elections are disabled",
1887                 behavior instanceof Leader);
1888     }
1889
1890     @Test
1891     public void testLaggingFollowerStarvation() throws Exception {
1892         logStart("testLaggingFollowerStarvation");
1893         new JavaTestKit(getSystem()) {{
1894             String leaderActorId = actorFactory.generateActorId("leader");
1895             String follower1ActorId = actorFactory.generateActorId("follower");
1896             String follower2ActorId = actorFactory.generateActorId("follower");
1897
1898             TestActorRef<ForwardMessageToBehaviorActor> leaderActor =
1899                     actorFactory.createTestActor(ForwardMessageToBehaviorActor.props(), leaderActorId);
1900             ActorRef follower1Actor = actorFactory.createActor(MessageCollectorActor.props(), follower1ActorId);
1901             ActorRef follower2Actor = actorFactory.createActor(MessageCollectorActor.props(), follower2ActorId);
1902
1903             MockRaftActorContext leaderActorContext =
1904                     new MockRaftActorContext(leaderActorId, getSystem(), leaderActor);
1905
1906             DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
1907             configParams.setHeartBeatInterval(new FiniteDuration(200, TimeUnit.MILLISECONDS));
1908             configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
1909
1910             leaderActorContext.setConfigParams(configParams);
1911
1912             leaderActorContext.setReplicatedLog(
1913                     new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(1,5,1).build());
1914
1915             Map<String, String> peerAddresses = new HashMap<>();
1916             peerAddresses.put(follower1ActorId,
1917                     follower1Actor.path().toString());
1918             peerAddresses.put(follower2ActorId,
1919                     follower2Actor.path().toString());
1920
1921             leaderActorContext.setPeerAddresses(peerAddresses);
1922             leaderActorContext.getTermInformation().update(1, leaderActorId);
1923
1924             RaftActorBehavior leader = createBehavior(leaderActorContext);
1925
1926             leaderActor.underlyingActor().setBehavior(leader);
1927
1928             for(int i=1;i<6;i++) {
1929                 // Each AppendEntriesReply could end up rescheduling the heartbeat (without the fix for bug 2733)
1930                 RaftActorBehavior newBehavior = leader.handleMessage(follower1Actor, new AppendEntriesReply(follower1ActorId, 1, true, i, 1, (short)0));
1931                 assertTrue(newBehavior == leader);
1932                 Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
1933             }
1934
1935             // Check if the leader has been receiving SendHeartbeat messages despite getting AppendEntriesReply
1936             List<SendHeartBeat> heartbeats = MessageCollectorActor.getAllMatching(leaderActor, SendHeartBeat.class);
1937
1938             assertTrue(String.format("%s heartbeat(s) is less than expected", heartbeats.size()),
1939                     heartbeats.size() > 1);
1940
1941             // Check if follower-2 got AppendEntries during this time and was not starved
1942             List<AppendEntries> appendEntries = MessageCollectorActor.getAllMatching(follower2Actor, AppendEntries.class);
1943
1944             assertTrue(String.format("%s append entries is less than expected", appendEntries.size()),
1945                     appendEntries.size() > 1);
1946
1947         }};
1948     }
1949
1950     @Test
1951     public void testReplicationConsensusWithNonVotingFollower() {
1952         logStart("testReplicationConsensusWithNonVotingFollower");
1953
1954         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1955         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1956                 new FiniteDuration(1000, TimeUnit.SECONDS));
1957
1958         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
1959
1960         String nonVotingFollowerId = "nonvoting-follower";
1961         TestActorRef<ForwardMessageToBehaviorActor> nonVotingFollowerActor = actorFactory.createTestActor(
1962                 Props.create(MessageCollectorActor.class), actorFactory.generateActorId(nonVotingFollowerId));
1963
1964         leaderActorContext.addToPeers(nonVotingFollowerId, nonVotingFollowerActor.path().toString(), VotingState.NON_VOTING);
1965
1966         leader = new Leader(leaderActorContext);
1967
1968         // Ignore initial heartbeats
1969         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1970         MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor, AppendEntries.class);
1971
1972         MessageCollectorActor.clearMessages(followerActor);
1973         MessageCollectorActor.clearMessages(nonVotingFollowerActor);
1974         MessageCollectorActor.clearMessages(leaderActor);
1975
1976         // Send a Replicate message and wait for AppendEntries.
1977         sendReplicate(leaderActorContext, 0);
1978
1979         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1980         MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor, AppendEntries.class);
1981
1982         // Send reply only from the voting follower and verify consensus via ApplyState.
1983         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 0, 1, (short)0));
1984
1985         MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class);
1986
1987         leader.handleMessage(leaderActor, new AppendEntriesReply(nonVotingFollowerId, 1, true, 0, 1, (short)0));
1988
1989         MessageCollectorActor.clearMessages(followerActor);
1990         MessageCollectorActor.clearMessages(nonVotingFollowerActor);
1991         MessageCollectorActor.clearMessages(leaderActor);
1992
1993         // Send another Replicate message
1994         sendReplicate(leaderActorContext, 1);
1995
1996         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1997         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor,
1998                 AppendEntries.class);
1999         assertEquals("Log entries size", 1, appendEntries.getEntries().size());
2000         assertEquals("Log entry index", 1, appendEntries.getEntries().get(0).getIndex());
2001
2002         // Send reply only from the non-voting follower and verify no consensus via no ApplyState.
2003         leader.handleMessage(leaderActor, new AppendEntriesReply(nonVotingFollowerId, 1, true, 1, 1, (short)0));
2004
2005         MessageCollectorActor.assertNoneMatching(leaderActor, ApplyState.class, 500);
2006
2007         // Send reply from the voting follower and verify consensus.
2008         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 1, 1, (short)0));
2009
2010         MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class);
2011     }
2012
2013     @Test
2014     public void testTransferLeadershipWithFollowerInSync() {
2015         logStart("testTransferLeadershipWithFollowerInSync");
2016
2017         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2018         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2019                 new FiniteDuration(1000, TimeUnit.SECONDS));
2020         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
2021
2022         leader = new Leader(leaderActorContext);
2023
2024         // Initial heartbeat
2025         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2026         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2027         MessageCollectorActor.clearMessages(followerActor);
2028
2029         sendReplicate(leaderActorContext, 0);
2030         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2031
2032         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 0, 1, (short)0));
2033         MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class);
2034         MessageCollectorActor.clearMessages(followerActor);
2035
2036         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2037         leader.transferLeadership(mockTransferCohort);
2038
2039         verify(mockTransferCohort, never()).transferComplete();
2040         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2041         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 0, 1, (short)0));
2042
2043         // Expect a final AppendEntries to ensure the follower's lastApplied index is up-to-date
2044         MessageCollectorActor.expectMatching(followerActor, AppendEntries.class, 2);
2045
2046         // Leader should force an election timeout
2047         MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class);
2048
2049         verify(mockTransferCohort).transferComplete();
2050     }
2051
2052     @Test
2053     public void testTransferLeadershipWithEmptyLog() {
2054         logStart("testTransferLeadershipWithEmptyLog");
2055
2056         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2057         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2058                 new FiniteDuration(1000, TimeUnit.SECONDS));
2059         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
2060
2061         leader = new Leader(leaderActorContext);
2062
2063         // Initial heartbeat
2064         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2065         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2066         MessageCollectorActor.clearMessages(followerActor);
2067
2068         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2069         leader.transferLeadership(mockTransferCohort);
2070
2071         verify(mockTransferCohort, never()).transferComplete();
2072         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2073         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2074
2075         // Expect a final AppendEntries to ensure the follower's lastApplied index is up-to-date
2076         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2077
2078         // Leader should force an election timeout
2079         MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class);
2080
2081         verify(mockTransferCohort).transferComplete();
2082     }
2083
2084     @Test
2085     public void testTransferLeadershipWithFollowerInitiallyOutOfSync() {
2086         logStart("testTransferLeadershipWithFollowerInitiallyOutOfSync");
2087
2088         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2089         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2090                 new FiniteDuration(200, TimeUnit.MILLISECONDS));
2091
2092         leader = new Leader(leaderActorContext);
2093
2094         // Initial heartbeat
2095         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2096         MessageCollectorActor.clearMessages(followerActor);
2097
2098         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2099         leader.transferLeadership(mockTransferCohort);
2100
2101         verify(mockTransferCohort, never()).transferComplete();
2102
2103         // Sync up the follower.
2104         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2105         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2106         MessageCollectorActor.clearMessages(followerActor);
2107
2108         Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().
2109                 getHeartBeatInterval().toMillis() + 1, TimeUnit.MILLISECONDS);
2110         leader.handleMessage(leaderActor, new SendHeartBeat());
2111         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2112         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 1, 1, (short)0));
2113
2114         // Leader should force an election timeout
2115         MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class);
2116
2117         verify(mockTransferCohort).transferComplete();
2118     }
2119
2120     @Test
2121     public void testTransferLeadershipWithFollowerSyncTimeout() {
2122         logStart("testTransferLeadershipWithFollowerSyncTimeout");
2123
2124         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2125         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2126                 new FiniteDuration(200, TimeUnit.MILLISECONDS));
2127         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(2);
2128         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
2129
2130         leader = new Leader(leaderActorContext);
2131
2132         // Initial heartbeat
2133         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2134         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2135         MessageCollectorActor.clearMessages(followerActor);
2136
2137         sendReplicate(leaderActorContext, 0);
2138         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2139
2140         MessageCollectorActor.clearMessages(followerActor);
2141
2142         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2143         leader.transferLeadership(mockTransferCohort);
2144
2145         verify(mockTransferCohort, never()).transferComplete();
2146
2147         // Send heartbeats to time out the transfer.
2148         for(int i = 0; i < leaderActorContext.getConfigParams().getElectionTimeoutFactor(); i++) {
2149             Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().
2150                     getHeartBeatInterval().toMillis() + 1, TimeUnit.MILLISECONDS);
2151             leader.handleMessage(leaderActor, new SendHeartBeat());
2152         }
2153
2154         verify(mockTransferCohort).abortTransfer();
2155         verify(mockTransferCohort, never()).transferComplete();
2156         MessageCollectorActor.assertNoneMatching(followerActor, ElectionTimeout.class, 100);
2157     }
2158
2159     @Override
2160     protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(RaftActorContext actorContext,
2161             ActorRef actorRef, RaftRPC rpc) throws Exception {
2162         super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
2163         assertEquals("New votedFor", null, actorContext.getTermInformation().getVotedFor());
2164     }
2165
2166     private class MockConfigParamsImpl extends DefaultConfigParamsImpl {
2167
2168         private final long electionTimeOutIntervalMillis;
2169         private final int snapshotChunkSize;
2170
2171         public MockConfigParamsImpl(long electionTimeOutIntervalMillis, int snapshotChunkSize) {
2172             super();
2173             this.electionTimeOutIntervalMillis = electionTimeOutIntervalMillis;
2174             this.snapshotChunkSize = snapshotChunkSize;
2175         }
2176
2177         @Override
2178         public FiniteDuration getElectionTimeOutInterval() {
2179             return new FiniteDuration(electionTimeOutIntervalMillis, TimeUnit.MILLISECONDS);
2180         }
2181
2182         @Override
2183         public int getSnapshotChunkSize() {
2184             return snapshotChunkSize;
2185         }
2186     }
2187 }