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

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