Cleanup warnings
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / behaviors / FollowerTest.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.Matchers.any;
18 import static org.mockito.Mockito.never;
19 import static org.mockito.Mockito.spy;
20 import static org.mockito.Mockito.verify;
21
22 import akka.actor.ActorRef;
23 import akka.actor.Props;
24 import akka.dispatch.Dispatchers;
25 import akka.testkit.JavaTestKit;
26 import akka.testkit.TestActorRef;
27 import com.google.common.base.Optional;
28 import com.google.common.base.Stopwatch;
29 import com.google.common.collect.ImmutableList;
30 import com.google.common.collect.ImmutableMap;
31 import com.google.common.io.ByteSource;
32 import com.google.common.util.concurrent.Uninterruptibles;
33 import com.google.protobuf.ByteString;
34 import java.io.OutputStream;
35 import java.util.ArrayList;
36 import java.util.Arrays;
37 import java.util.Collections;
38 import java.util.HashMap;
39 import java.util.List;
40 import java.util.concurrent.TimeUnit;
41 import java.util.concurrent.atomic.AtomicReference;
42 import org.junit.After;
43 import org.junit.Assert;
44 import org.junit.Test;
45 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
46 import org.opendaylight.controller.cluster.raft.MockRaftActor;
47 import org.opendaylight.controller.cluster.raft.MockRaftActor.Builder;
48 import org.opendaylight.controller.cluster.raft.MockRaftActor.MockSnapshotState;
49 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
50 import org.opendaylight.controller.cluster.raft.RaftActorContext;
51 import org.opendaylight.controller.cluster.raft.RaftActorSnapshotCohort;
52 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
53 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
54 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
55 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
56 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
57 import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
58 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
59 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
60 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
61 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
62 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
63 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
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.ServerConfigurationPayload;
68 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
69 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
70 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
71 import org.opendaylight.controller.cluster.raft.persisted.Snapshot.State;
72 import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
73 import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
74 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
75 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
76 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
77 import scala.concurrent.duration.FiniteDuration;
78
79 public class FollowerTest extends AbstractRaftActorBehaviorTest<Follower> {
80
81     private final TestActorRef<MessageCollectorActor> followerActor = actorFactory.createTestActor(
82             Props.create(MessageCollectorActor.class), actorFactory.generateActorId("follower"));
83
84     private final TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
85             Props.create(MessageCollectorActor.class), actorFactory.generateActorId("leader"));
86
87     private Follower follower;
88
89     private final short payloadVersion = 5;
90
91     @Override
92     @After
93     public void tearDown() {
94         if (follower != null) {
95             follower.close();
96         }
97
98         super.tearDown();
99     }
100
101     @Override
102     protected Follower createBehavior(final RaftActorContext actorContext) {
103         return spy(new Follower(actorContext));
104     }
105
106     @Override
107     protected  MockRaftActorContext createActorContext() {
108         return createActorContext(followerActor);
109     }
110
111     @Override
112     protected  MockRaftActorContext createActorContext(final ActorRef actorRef) {
113         MockRaftActorContext context = new MockRaftActorContext("follower", getSystem(), actorRef);
114         context.setPayloadVersion(payloadVersion);
115         return context;
116     }
117
118     @Test
119     public void testThatAnElectionTimeoutIsTriggered() {
120         MockRaftActorContext actorContext = createActorContext();
121         follower = new Follower(actorContext);
122
123         MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class,
124                 actorContext.getConfigParams().getElectionTimeOutInterval().$times(6).toMillis());
125     }
126
127     @Test
128     public void testHandleElectionTimeoutWhenNoLeaderMessageReceived() {
129         logStart("testHandleElectionTimeoutWhenNoLeaderMessageReceived");
130
131         MockRaftActorContext context = createActorContext();
132         follower = new Follower(context);
133
134         Uninterruptibles.sleepUninterruptibly(context.getConfigParams().getElectionTimeOutInterval().toMillis(),
135                 TimeUnit.MILLISECONDS);
136         RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
137
138         assertTrue(raftBehavior instanceof Candidate);
139     }
140
141     @Test
142     public void testHandleElectionTimeoutWhenLeaderMessageReceived() {
143         logStart("testHandleElectionTimeoutWhenLeaderMessageReceived");
144
145         MockRaftActorContext context = createActorContext();
146         ((DefaultConfigParamsImpl) context.getConfigParams())
147                 .setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
148         ((DefaultConfigParamsImpl) context.getConfigParams()).setElectionTimeoutFactor(4);
149
150         follower = new Follower(context);
151         context.setCurrentBehavior(follower);
152
153         Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
154                 .getElectionTimeOutInterval().toMillis() - 100, TimeUnit.MILLISECONDS);
155         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
156                 -1, -1, (short) 1));
157
158         Uninterruptibles.sleepUninterruptibly(130, TimeUnit.MILLISECONDS);
159         RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
160         assertTrue(raftBehavior instanceof Follower);
161
162         Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
163                 .getElectionTimeOutInterval().toMillis() - 150, TimeUnit.MILLISECONDS);
164         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
165                 -1, -1, (short) 1));
166
167         Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
168         raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
169         assertTrue(raftBehavior instanceof Follower);
170     }
171
172     @Test
173     public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull() {
174         logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull");
175
176         MockRaftActorContext context = createActorContext();
177         long term = 1000;
178         context.getTermInformation().update(term, null);
179
180         follower = createBehavior(context);
181
182         follower.handleMessage(leaderActor, new RequestVote(term, "test", 10000, 999));
183
184         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
185
186         assertEquals("isVoteGranted", true, reply.isVoteGranted());
187         assertEquals("getTerm", term, reply.getTerm());
188         verify(follower).scheduleElection(any(FiniteDuration.class));
189     }
190
191     @Test
192     public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId() {
193         logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId");
194
195         MockRaftActorContext context = createActorContext();
196         long term = 1000;
197         context.getTermInformation().update(term, "test");
198
199         follower = createBehavior(context);
200
201         follower.handleMessage(leaderActor, new RequestVote(term, "candidate", 10000, 999));
202
203         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
204
205         assertEquals("isVoteGranted", false, reply.isVoteGranted());
206         verify(follower, never()).scheduleElection(any(FiniteDuration.class));
207     }
208
209
210     @Test
211     public void testHandleFirstAppendEntries() {
212         logStart("testHandleFirstAppendEntries");
213
214         MockRaftActorContext context = createActorContext();
215         context.getReplicatedLog().clear(0,2);
216         context.getReplicatedLog().append(newReplicatedLogEntry(1,100, "bar"));
217         context.getReplicatedLog().setSnapshotIndex(99);
218
219         List<ReplicatedLogEntry> entries = Arrays.asList(
220                 newReplicatedLogEntry(2, 101, "foo"));
221
222         Assert.assertEquals(1, context.getReplicatedLog().size());
223
224         // The new commitIndex is 101
225         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
226
227         follower = createBehavior(context);
228         follower.handleMessage(leaderActor, appendEntries);
229
230         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
231                 FollowerInitialSyncUpStatus.class);
232         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
233
234         assertFalse(syncStatus.isInitialSyncDone());
235         assertTrue("append entries reply should be true", reply.isSuccess());
236     }
237
238     @Test
239     public void testHandleFirstAppendEntriesWithPrevIndexMinusOne() {
240         logStart("testHandleFirstAppendEntries");
241
242         MockRaftActorContext context = createActorContext();
243
244         List<ReplicatedLogEntry> entries = Arrays.asList(
245                 newReplicatedLogEntry(2, 101, "foo"));
246
247         // The new commitIndex is 101
248         AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
249
250         follower = createBehavior(context);
251         follower.handleMessage(leaderActor, appendEntries);
252
253         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
254                 FollowerInitialSyncUpStatus.class);
255         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
256
257         assertFalse(syncStatus.isInitialSyncDone());
258         assertFalse("append entries reply should be false", reply.isSuccess());
259     }
260
261     @Test
262     public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog() {
263         logStart("testHandleFirstAppendEntries");
264
265         MockRaftActorContext context = createActorContext();
266         context.getReplicatedLog().clear(0,2);
267         context.getReplicatedLog().append(newReplicatedLogEntry(1, 100, "bar"));
268         context.getReplicatedLog().setSnapshotIndex(99);
269
270         List<ReplicatedLogEntry> entries = Arrays.asList(
271                 newReplicatedLogEntry(2, 101, "foo"));
272
273         // The new commitIndex is 101
274         AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
275
276         follower = createBehavior(context);
277         follower.handleMessage(leaderActor, appendEntries);
278
279         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
280                 FollowerInitialSyncUpStatus.class);
281         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
282
283         assertFalse(syncStatus.isInitialSyncDone());
284         assertTrue("append entries reply should be true", reply.isSuccess());
285     }
286
287     @Test
288     public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot() {
289         logStart("testHandleFirstAppendEntries");
290
291         MockRaftActorContext context = createActorContext();
292         context.getReplicatedLog().clear(0,2);
293         context.getReplicatedLog().setSnapshotIndex(100);
294
295         List<ReplicatedLogEntry> entries = Arrays.asList(
296                 newReplicatedLogEntry(2, 101, "foo"));
297
298         // The new commitIndex is 101
299         AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
300
301         follower = createBehavior(context);
302         follower.handleMessage(leaderActor, appendEntries);
303
304         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
305                 FollowerInitialSyncUpStatus.class);
306         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
307
308         assertFalse(syncStatus.isInitialSyncDone());
309         assertTrue("append entries reply should be true", reply.isSuccess());
310     }
311
312     @Test
313     public void testFirstAppendEntriesWithNoPrevIndexAndReplToAllPresentInSnapshotButCalculatedPrevEntryMissing() {
314         logStart(
315                "testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing");
316
317         MockRaftActorContext context = createActorContext();
318         context.getReplicatedLog().clear(0,2);
319         context.getReplicatedLog().setSnapshotIndex(100);
320
321         List<ReplicatedLogEntry> entries = Arrays.asList(
322                 newReplicatedLogEntry(2, 105, "foo"));
323
324         // The new commitIndex is 101
325         AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 105, 100, (short) 0);
326
327         follower = createBehavior(context);
328         follower.handleMessage(leaderActor, appendEntries);
329
330         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
331                 FollowerInitialSyncUpStatus.class);
332         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
333
334         assertFalse(syncStatus.isInitialSyncDone());
335         assertFalse("append entries reply should be false", reply.isSuccess());
336     }
337
338     @Test
339     public void testHandleSyncUpAppendEntries() {
340         logStart("testHandleSyncUpAppendEntries");
341
342         MockRaftActorContext context = createActorContext();
343
344         List<ReplicatedLogEntry> entries = Arrays.asList(
345                 newReplicatedLogEntry(2, 101, "foo"));
346
347         // The new commitIndex is 101
348         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
349
350         follower = createBehavior(context);
351         follower.handleMessage(leaderActor, appendEntries);
352
353         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
354                 FollowerInitialSyncUpStatus.class);
355
356         assertFalse(syncStatus.isInitialSyncDone());
357
358         // Clear all the messages
359         followerActor.underlyingActor().clear();
360
361         context.setLastApplied(101);
362         context.setCommitIndex(101);
363         setLastLogEntry(context, 1, 101, new MockRaftActorContext.MockPayload(""));
364
365         entries = Arrays.asList(newReplicatedLogEntry(2, 101, "foo"));
366
367         // The new commitIndex is 101
368         appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0);
369         follower.handleMessage(leaderActor, appendEntries);
370
371         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
372
373         assertTrue(syncStatus.isInitialSyncDone());
374
375         followerActor.underlyingActor().clear();
376
377         // Sending the same message again should not generate another message
378
379         follower.handleMessage(leaderActor, appendEntries);
380
381         syncStatus = MessageCollectorActor.getFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
382
383         assertNull(syncStatus);
384     }
385
386     @Test
387     public void testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete() {
388         logStart("testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete");
389
390         MockRaftActorContext context = createActorContext();
391
392         List<ReplicatedLogEntry> entries = Arrays.asList(
393                 newReplicatedLogEntry(2, 101, "foo"));
394
395         // The new commitIndex is 101
396         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
397
398         follower = createBehavior(context);
399         follower.handleMessage(leaderActor, appendEntries);
400
401         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
402                 FollowerInitialSyncUpStatus.class);
403
404         assertFalse(syncStatus.isInitialSyncDone());
405
406         // Clear all the messages
407         followerActor.underlyingActor().clear();
408
409         context.setLastApplied(100);
410         setLastLogEntry(context, 1, 100,
411                 new MockRaftActorContext.MockPayload(""));
412
413         entries = Arrays.asList(
414                 newReplicatedLogEntry(2, 101, "foo"));
415
416         // leader-2 is becoming the leader now and it says the commitIndex is 45
417         appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0);
418         follower.handleMessage(leaderActor, appendEntries);
419
420         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
421
422         // We get a new message saying initial status is not done
423         assertFalse(syncStatus.isInitialSyncDone());
424     }
425
426     @Test
427     public void testHandleAppendEntriesLeaderChangedAfterSyncUpComplete() {
428         logStart("testHandleAppendEntriesLeaderChangedAfterSyncUpComplete");
429
430         MockRaftActorContext context = createActorContext();
431
432         List<ReplicatedLogEntry> entries = Arrays.asList(
433                 newReplicatedLogEntry(2, 101, "foo"));
434
435         // The new commitIndex is 101
436         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
437
438         follower = createBehavior(context);
439         follower.handleMessage(leaderActor, appendEntries);
440
441         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
442                 FollowerInitialSyncUpStatus.class);
443
444         assertFalse(syncStatus.isInitialSyncDone());
445
446         // Clear all the messages
447         followerActor.underlyingActor().clear();
448
449         context.setLastApplied(101);
450         context.setCommitIndex(101);
451         setLastLogEntry(context, 1, 101,
452                 new MockRaftActorContext.MockPayload(""));
453
454         entries = Arrays.asList(
455                 newReplicatedLogEntry(2, 101, "foo"));
456
457         // The new commitIndex is 101
458         appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0);
459         follower.handleMessage(leaderActor, appendEntries);
460
461         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
462
463         assertTrue(syncStatus.isInitialSyncDone());
464
465         // Clear all the messages
466         followerActor.underlyingActor().clear();
467
468         context.setLastApplied(100);
469         setLastLogEntry(context, 1, 100,
470                 new MockRaftActorContext.MockPayload(""));
471
472         entries = Arrays.asList(
473                 newReplicatedLogEntry(2, 101, "foo"));
474
475         // leader-2 is becoming the leader now and it says the commitIndex is 45
476         appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0);
477         follower.handleMessage(leaderActor, appendEntries);
478
479         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
480
481         // We get a new message saying initial status is not done
482         assertFalse(syncStatus.isInitialSyncDone());
483     }
484
485     /**
486      * This test verifies that when an AppendEntries RPC is received by a RaftActor
487      * with a commitIndex that is greater than what has been applied to the
488      * state machine of the RaftActor, the RaftActor applies the state and
489      * sets it current applied state to the commitIndex of the sender.
490      */
491     @Test
492     public void testHandleAppendEntriesWithNewerCommitIndex() {
493         logStart("testHandleAppendEntriesWithNewerCommitIndex");
494
495         MockRaftActorContext context = createActorContext();
496
497         context.setLastApplied(100);
498         setLastLogEntry(context, 1, 100,
499                 new MockRaftActorContext.MockPayload(""));
500         context.getReplicatedLog().setSnapshotIndex(99);
501
502         List<ReplicatedLogEntry> entries = Arrays.<ReplicatedLogEntry>asList(
503                 newReplicatedLogEntry(2, 101, "foo"));
504
505         // The new commitIndex is 101
506         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
507
508         follower = createBehavior(context);
509         follower.handleMessage(leaderActor, appendEntries);
510
511         assertEquals("getLastApplied", 101L, context.getLastApplied());
512     }
513
514     /**
515      * This test verifies that when an AppendEntries is received a specific prevLogTerm
516      * which does not match the term that is in RaftActors log entry at prevLogIndex
517      * then the RaftActor does not change it's state and it returns a failure.
518      */
519     @Test
520     public void testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() {
521         logStart("testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm");
522
523         MockRaftActorContext context = createActorContext();
524
525         // First set the receivers term to lower number
526         context.getTermInformation().update(95, "test");
527
528         // AppendEntries is now sent with a bigger term
529         // this will set the receivers term to be the same as the sender's term
530         AppendEntries appendEntries = new AppendEntries(100, "leader", 0, 0, Collections.emptyList(), 101, -1,
531                 (short)0);
532
533         follower = createBehavior(context);
534
535         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
536
537         Assert.assertSame(follower, newBehavior);
538
539         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
540                 AppendEntriesReply.class);
541
542         assertEquals("isSuccess", false, reply.isSuccess());
543     }
544
545     /**
546      * This test verifies that when a new AppendEntries message is received with
547      * new entries and the logs of the sender and receiver match that the new
548      * entries get added to the log and the log is incremented by the number of
549      * entries received in appendEntries.
550      */
551     @Test
552     public void testHandleAppendEntriesAddNewEntries() {
553         logStart("testHandleAppendEntriesAddNewEntries");
554
555         MockRaftActorContext context = createActorContext();
556
557         // First set the receivers term to lower number
558         context.getTermInformation().update(1, "test");
559
560         // Prepare the receivers log
561         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
562         log.append(newReplicatedLogEntry(1, 0, "zero"));
563         log.append(newReplicatedLogEntry(1, 1, "one"));
564         log.append(newReplicatedLogEntry(1, 2, "two"));
565
566         context.setReplicatedLog(log);
567
568         // Prepare the entries to be sent with AppendEntries
569         List<ReplicatedLogEntry> entries = new ArrayList<>();
570         entries.add(newReplicatedLogEntry(1, 3, "three"));
571         entries.add(newReplicatedLogEntry(1, 4, "four"));
572
573         // Send appendEntries with the same term as was set on the receiver
574         // before the new behavior was created (1 in this case)
575         // This will not work for a Candidate because as soon as a Candidate
576         // is created it increments the term
577         short leaderPayloadVersion = 10;
578         String leaderId = "leader-1";
579         AppendEntries appendEntries = new AppendEntries(1, leaderId, 2, 1, entries, 4, -1, leaderPayloadVersion);
580
581         follower = createBehavior(context);
582
583         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
584
585         Assert.assertSame(follower, newBehavior);
586
587         assertEquals("Next index", 5, log.last().getIndex() + 1);
588         assertEquals("Entry 3", entries.get(0), log.get(3));
589         assertEquals("Entry 4", entries.get(1), log.get(4));
590
591         assertEquals("getLeaderPayloadVersion", leaderPayloadVersion, newBehavior.getLeaderPayloadVersion());
592         assertEquals("getLeaderId", leaderId, newBehavior.getLeaderId());
593
594         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
595     }
596
597     /**
598      * This test verifies that when a new AppendEntries message is received with
599      * new entries and the logs of the sender and receiver are out-of-sync that
600      * the log is first corrected by removing the out of sync entries from the
601      * log and then adding in the new entries sent with the AppendEntries message.
602      */
603     @Test
604     public void testHandleAppendEntriesCorrectReceiverLogEntries() {
605         logStart("testHandleAppendEntriesCorrectReceiverLogEntries");
606
607         MockRaftActorContext context = createActorContext();
608
609         // First set the receivers term to lower number
610         context.getTermInformation().update(1, "test");
611
612         // Prepare the receivers log
613         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
614         log.append(newReplicatedLogEntry(1, 0, "zero"));
615         log.append(newReplicatedLogEntry(1, 1, "one"));
616         log.append(newReplicatedLogEntry(1, 2, "two"));
617
618         context.setReplicatedLog(log);
619
620         // Prepare the entries to be sent with AppendEntries
621         List<ReplicatedLogEntry> entries = new ArrayList<>();
622         entries.add(newReplicatedLogEntry(2, 2, "two-1"));
623         entries.add(newReplicatedLogEntry(2, 3, "three"));
624
625         // Send appendEntries with the same term as was set on the receiver
626         // before the new behavior was created (1 in this case)
627         // This will not work for a Candidate because as soon as a Candidate
628         // is created it increments the term
629         AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1, (short)0);
630
631         follower = createBehavior(context);
632
633         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
634
635         Assert.assertSame(follower, newBehavior);
636
637         // The entry at index 2 will be found out-of-sync with the leader
638         // and will be removed
639         // Then the two new entries will be added to the log
640         // Thus making the log to have 4 entries
641         assertEquals("Next index", 4, log.last().getIndex() + 1);
642         //assertEquals("Entry 2", entries.get(0), log.get(2));
643
644         assertEquals("Entry 1 data", "one", log.get(1).getData().toString());
645
646         // Check that the entry at index 2 has the new data
647         assertEquals("Entry 2", entries.get(0), log.get(2));
648
649         assertEquals("Entry 3", entries.get(1), log.get(3));
650
651         expectAndVerifyAppendEntriesReply(2, true, context.getId(), 2, 3);
652     }
653
654     @Test
655     public void testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot() {
656         logStart("testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot");
657
658         MockRaftActorContext context = createActorContext();
659
660         // First set the receivers term to lower number
661         context.getTermInformation().update(1, "test");
662
663         // Prepare the receivers log
664         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
665         log.append(newReplicatedLogEntry(1, 0, "zero"));
666         log.append(newReplicatedLogEntry(1, 1, "one"));
667         log.append(newReplicatedLogEntry(1, 2, "two"));
668
669         context.setReplicatedLog(log);
670
671         // Prepare the entries to be sent with AppendEntries
672         List<ReplicatedLogEntry> entries = new ArrayList<>();
673         entries.add(newReplicatedLogEntry(2, 2, "two-1"));
674         entries.add(newReplicatedLogEntry(2, 3, "three"));
675
676         // Send appendEntries with the same term as was set on the receiver
677         // before the new behavior was created (1 in this case)
678         // This will not work for a Candidate because as soon as a Candidate
679         // is created it increments the term
680         AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1, (short)0);
681
682         context.setRaftPolicy(createRaftPolicy(false, true));
683         follower = createBehavior(context);
684
685         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
686
687         Assert.assertSame(follower, newBehavior);
688
689         expectAndVerifyAppendEntriesReply(2, false, context.getId(), 1, 2, true);
690     }
691
692     @Test
693     public void testHandleAppendEntriesPreviousLogEntryMissing() {
694         logStart("testHandleAppendEntriesPreviousLogEntryMissing");
695
696         final MockRaftActorContext context = createActorContext();
697
698         // Prepare the receivers log
699         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
700         log.append(newReplicatedLogEntry(1, 0, "zero"));
701         log.append(newReplicatedLogEntry(1, 1, "one"));
702         log.append(newReplicatedLogEntry(1, 2, "two"));
703
704         context.setReplicatedLog(log);
705
706         // Prepare the entries to be sent with AppendEntries
707         List<ReplicatedLogEntry> entries = new ArrayList<>();
708         entries.add(newReplicatedLogEntry(1, 4, "four"));
709
710         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1, (short)0);
711
712         follower = createBehavior(context);
713
714         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
715
716         Assert.assertSame(follower, newBehavior);
717
718         expectAndVerifyAppendEntriesReply(1, false, context.getId(), 1, 2);
719     }
720
721     @Test
722     public void testHandleAppendEntriesWithExistingLogEntry() {
723         logStart("testHandleAppendEntriesWithExistingLogEntry");
724
725         MockRaftActorContext context = createActorContext();
726
727         context.getTermInformation().update(1, "test");
728
729         // Prepare the receivers log
730         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
731         log.append(newReplicatedLogEntry(1, 0, "zero"));
732         log.append(newReplicatedLogEntry(1, 1, "one"));
733
734         context.setReplicatedLog(log);
735
736         // Send the last entry again.
737         List<ReplicatedLogEntry> entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"));
738
739         follower = createBehavior(context);
740
741         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1, (short)0));
742
743         assertEquals("Next index", 2, log.last().getIndex() + 1);
744         assertEquals("Entry 1", entries.get(0), log.get(1));
745
746         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 1);
747
748         // Send the last entry again and also a new one.
749
750         entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two"));
751
752         leaderActor.underlyingActor().clear();
753         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1, (short)0));
754
755         assertEquals("Next index", 3, log.last().getIndex() + 1);
756         assertEquals("Entry 1", entries.get(0), log.get(1));
757         assertEquals("Entry 2", entries.get(1), log.get(2));
758
759         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 2);
760     }
761
762     @Test
763     public void testHandleAppendEntriesAfterInstallingSnapshot() {
764         logStart("testHandleAppendAfterInstallingSnapshot");
765
766         MockRaftActorContext context = createActorContext();
767
768         // Prepare the receivers log
769         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
770
771         // Set up a log as if it has been snapshotted
772         log.setSnapshotIndex(3);
773         log.setSnapshotTerm(1);
774
775         context.setReplicatedLog(log);
776
777         // Prepare the entries to be sent with AppendEntries
778         List<ReplicatedLogEntry> entries = new ArrayList<>();
779         entries.add(newReplicatedLogEntry(1, 4, "four"));
780
781         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3, (short)0);
782
783         follower = createBehavior(context);
784
785         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
786
787         Assert.assertSame(follower, newBehavior);
788
789         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
790     }
791
792
793     /**
794      * This test verifies that when InstallSnapshot is received by
795      * the follower its applied correctly.
796      */
797     @Test
798     public void testHandleInstallSnapshot() {
799         logStart("testHandleInstallSnapshot");
800
801         MockRaftActorContext context = createActorContext();
802         context.getTermInformation().update(1, "leader");
803
804         follower = createBehavior(context);
805
806         ByteString bsSnapshot = createSnapshot();
807         int offset = 0;
808         int snapshotLength = bsSnapshot.size();
809         int chunkSize = 50;
810         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
811         int lastIncludedIndex = 1;
812         int chunkIndex = 1;
813         InstallSnapshot lastInstallSnapshot = null;
814
815         for (int i = 0; i < totalChunks; i++) {
816             byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
817             lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
818                     chunkData, chunkIndex, totalChunks);
819             follower.handleMessage(leaderActor, lastInstallSnapshot);
820             offset = offset + 50;
821             lastIncludedIndex++;
822             chunkIndex++;
823         }
824
825         ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
826                 ApplySnapshot.class);
827         Snapshot snapshot = applySnapshot.getSnapshot();
828         assertNotNull(lastInstallSnapshot);
829         assertEquals("getLastIndex", lastInstallSnapshot.getLastIncludedIndex(), snapshot.getLastIndex());
830         assertEquals("getLastIncludedTerm", lastInstallSnapshot.getLastIncludedTerm(),
831                 snapshot.getLastAppliedTerm());
832         assertEquals("getLastAppliedIndex", lastInstallSnapshot.getLastIncludedIndex(),
833                 snapshot.getLastAppliedIndex());
834         assertEquals("getLastTerm", lastInstallSnapshot.getLastIncludedTerm(), snapshot.getLastTerm());
835         assertEquals("getState type", ByteState.class, snapshot.getState().getClass());
836         Assert.assertArrayEquals("getState", bsSnapshot.toByteArray(), ((ByteState)snapshot.getState()).getBytes());
837         assertEquals("getElectionTerm", 1, snapshot.getElectionTerm());
838         assertEquals("getElectionVotedFor", "leader", snapshot.getElectionVotedFor());
839         applySnapshot.getCallback().onSuccess();
840
841         List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
842                 leaderActor, InstallSnapshotReply.class);
843         assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
844
845         chunkIndex = 1;
846         for (InstallSnapshotReply reply: replies) {
847             assertEquals("getChunkIndex", chunkIndex++, reply.getChunkIndex());
848             assertEquals("getTerm", 1, reply.getTerm());
849             assertEquals("isSuccess", true, reply.isSuccess());
850             assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
851         }
852
853         assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
854     }
855
856     /**
857      * Verify that when an AppendEntries is sent to a follower during a snapshot install
858      * the Follower short-circuits the processing of the AppendEntries message.
859      */
860     @Test
861     public void testReceivingAppendEntriesDuringInstallSnapshot() {
862         logStart("testReceivingAppendEntriesDuringInstallSnapshot");
863
864         MockRaftActorContext context = createActorContext();
865
866         follower = createBehavior(context);
867
868         ByteString bsSnapshot  = createSnapshot();
869         int snapshotLength = bsSnapshot.size();
870         int chunkSize = 50;
871         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
872         int lastIncludedIndex = 1;
873
874         // Check that snapshot installation is not in progress
875         assertNull(follower.getSnapshotTracker());
876
877         // Make sure that we have more than 1 chunk to send
878         assertTrue(totalChunks > 1);
879
880         // Send an install snapshot with the first chunk to start the process of installing a snapshot
881         byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
882         follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
883                 chunkData, 1, totalChunks));
884
885         // Check if snapshot installation is in progress now
886         assertNotNull(follower.getSnapshotTracker());
887
888         // Send an append entry
889         AppendEntries appendEntries = new AppendEntries(1, "leader", 1, 1,
890                 Arrays.asList(newReplicatedLogEntry(2, 1, "3")), 2, -1, (short)1);
891
892         follower.handleMessage(leaderActor, appendEntries);
893
894         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
895         assertEquals("isSuccess", true, reply.isSuccess());
896         assertEquals("getLogLastIndex", context.getReplicatedLog().lastIndex(), reply.getLogLastIndex());
897         assertEquals("getLogLastTerm", context.getReplicatedLog().lastTerm(), reply.getLogLastTerm());
898         assertEquals("getTerm", context.getTermInformation().getCurrentTerm(), reply.getTerm());
899
900         assertNotNull(follower.getSnapshotTracker());
901     }
902
903     @Test
904     public void testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader() {
905         logStart("testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader");
906
907         MockRaftActorContext context = createActorContext();
908
909         follower = createBehavior(context);
910
911         ByteString bsSnapshot  = createSnapshot();
912         int snapshotLength = bsSnapshot.size();
913         int chunkSize = 50;
914         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
915         int lastIncludedIndex = 1;
916
917         // Check that snapshot installation is not in progress
918         assertNull(follower.getSnapshotTracker());
919
920         // Make sure that we have more than 1 chunk to send
921         assertTrue(totalChunks > 1);
922
923         // Send an install snapshot with the first chunk to start the process of installing a snapshot
924         byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
925         follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
926                 chunkData, 1, totalChunks));
927
928         // Check if snapshot installation is in progress now
929         assertNotNull(follower.getSnapshotTracker());
930
931         // Send appendEntries with a new term and leader.
932         AppendEntries appendEntries = new AppendEntries(2, "new-leader", 1, 1,
933                 Arrays.asList(newReplicatedLogEntry(2, 2, "3")), 2, -1, (short)1);
934
935         follower.handleMessage(leaderActor, appendEntries);
936
937         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
938         assertEquals("isSuccess", true, reply.isSuccess());
939         assertEquals("getLogLastIndex", 2, reply.getLogLastIndex());
940         assertEquals("getLogLastTerm", 2, reply.getLogLastTerm());
941         assertEquals("getTerm", 2, reply.getTerm());
942
943         assertNull(follower.getSnapshotTracker());
944     }
945
946     @Test
947     public void testInitialSyncUpWithHandleInstallSnapshotFollowedByAppendEntries() {
948         logStart("testInitialSyncUpWithHandleInstallSnapshot");
949
950         MockRaftActorContext context = createActorContext();
951         context.setCommitIndex(-1);
952
953         follower = createBehavior(context);
954
955         ByteString bsSnapshot  = createSnapshot();
956         int offset = 0;
957         int snapshotLength = bsSnapshot.size();
958         int chunkSize = 50;
959         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
960         int lastIncludedIndex = 1;
961         int chunkIndex = 1;
962         InstallSnapshot lastInstallSnapshot = null;
963
964         for (int i = 0; i < totalChunks; i++) {
965             byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
966             lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
967                     chunkData, chunkIndex, totalChunks);
968             follower.handleMessage(leaderActor, lastInstallSnapshot);
969             offset = offset + 50;
970             lastIncludedIndex++;
971             chunkIndex++;
972         }
973
974         FollowerInitialSyncUpStatus syncStatus =
975                 MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
976
977         assertFalse(syncStatus.isInitialSyncDone());
978
979         // Clear all the messages
980         followerActor.underlyingActor().clear();
981
982         context.setLastApplied(101);
983         context.setCommitIndex(101);
984         setLastLogEntry(context, 1, 101,
985                 new MockRaftActorContext.MockPayload(""));
986
987         List<ReplicatedLogEntry> entries = Arrays.asList(
988                 newReplicatedLogEntry(2, 101, "foo"));
989
990         // The new commitIndex is 101
991         AppendEntries appendEntries = new AppendEntries(2, "leader", 101, 1, entries, 102, 101, (short)0);
992         follower.handleMessage(leaderActor, appendEntries);
993
994         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
995
996         assertTrue(syncStatus.isInitialSyncDone());
997     }
998
999     @Test
1000     public void testHandleOutOfSequenceInstallSnapshot() {
1001         logStart("testHandleOutOfSequenceInstallSnapshot");
1002
1003         MockRaftActorContext context = createActorContext();
1004
1005         follower = createBehavior(context);
1006
1007         ByteString bsSnapshot = createSnapshot();
1008
1009         InstallSnapshot installSnapshot = new InstallSnapshot(1, "leader", 3, 1,
1010                 getNextChunk(bsSnapshot, 10, 50), 3, 3);
1011         follower.handleMessage(leaderActor, installSnapshot);
1012
1013         InstallSnapshotReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
1014                 InstallSnapshotReply.class);
1015
1016         assertEquals("isSuccess", false, reply.isSuccess());
1017         assertEquals("getChunkIndex", -1, reply.getChunkIndex());
1018         assertEquals("getTerm", 1, reply.getTerm());
1019         assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
1020
1021         assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
1022     }
1023
1024     @Test
1025     public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers() {
1026         MockRaftActorContext context = createActorContext();
1027
1028         Stopwatch stopwatch = Stopwatch.createStarted();
1029
1030         follower = createBehavior(context);
1031
1032         TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
1033
1034         long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
1035
1036         assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis());
1037
1038         RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
1039         assertTrue("Expected Candidate", newBehavior instanceof Candidate);
1040     }
1041
1042     @Test
1043     public void testFollowerSchedulesElectionIfAutomaticElectionsAreDisabled() {
1044         MockRaftActorContext context = createActorContext();
1045         context.setConfigParams(new DefaultConfigParamsImpl() {
1046             @Override
1047             public FiniteDuration getElectionTimeOutInterval() {
1048                 return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
1049             }
1050         });
1051
1052         context.setRaftPolicy(createRaftPolicy(false, false));
1053
1054         follower = createBehavior(context);
1055
1056         TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
1057         RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
1058         assertSame("handleMessage result", follower, newBehavior);
1059     }
1060
1061     @Test
1062     public void testFollowerSchedulesElectionIfNonVoting() {
1063         MockRaftActorContext context = createActorContext();
1064         context.updatePeerIds(new ServerConfigurationPayload(Arrays.asList(new ServerInfo(context.getId(), false))));
1065         ((DefaultConfigParamsImpl)context.getConfigParams()).setHeartBeatInterval(
1066                 FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
1067         ((DefaultConfigParamsImpl)context.getConfigParams()).setElectionTimeoutFactor(1);
1068
1069         follower = new Follower(context, "leader", (short)1);
1070
1071         ElectionTimeout electionTimeout = MessageCollectorActor.expectFirstMatching(followerActor,
1072                 ElectionTimeout.class);
1073         RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), electionTimeout);
1074         assertSame("handleMessage result", follower, newBehavior);
1075         assertNull("Expected null leaderId", follower.getLeaderId());
1076     }
1077
1078     @Test
1079     public void testElectionScheduledWhenAnyRaftRPCReceived() {
1080         MockRaftActorContext context = createActorContext();
1081         follower = createBehavior(context);
1082         follower.handleMessage(leaderActor, new RaftRPC() {
1083             private static final long serialVersionUID = 1L;
1084
1085             @Override
1086             public long getTerm() {
1087                 return 100;
1088             }
1089         });
1090         verify(follower).scheduleElection(any(FiniteDuration.class));
1091     }
1092
1093     @Test
1094     public void testElectionNotScheduledWhenNonRaftRPCMessageReceived() {
1095         MockRaftActorContext context = createActorContext();
1096         follower = createBehavior(context);
1097         follower.handleMessage(leaderActor, "non-raft-rpc");
1098         verify(follower, never()).scheduleElection(any(FiniteDuration.class));
1099     }
1100
1101     @Test
1102     public void testCaptureSnapshotOnLastEntryInAppendEntries() {
1103         String id = "testCaptureSnapshotOnLastEntryInAppendEntries";
1104         logStart(id);
1105
1106         InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1107
1108         DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1109         config.setSnapshotBatchCount(2);
1110         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1111
1112         final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
1113         RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
1114         Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
1115                 .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
1116         TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
1117                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1118         followerRaftActor.set(followerActorRef.underlyingActor());
1119         followerRaftActor.get().waitForInitializeBehaviorComplete();
1120
1121         InMemorySnapshotStore.addSnapshotSavedLatch(id);
1122         InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1123         InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
1124
1125         List<ReplicatedLogEntry> entries = Arrays.asList(
1126                 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"));
1127
1128         AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 1, -1, (short)0);
1129
1130         followerActorRef.tell(appendEntries, leaderActor);
1131
1132         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1133         assertEquals("isSuccess", true, reply.isSuccess());
1134
1135         final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
1136
1137         InMemoryJournal.waitForDeleteMessagesComplete(id);
1138         InMemoryJournal.waitForWriteMessagesComplete(id);
1139         // We expect the ApplyJournalEntries for index 1 to remain in the persisted log b/c it's still queued for
1140         // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
1141         // This is OK - on recovery it will be a no-op since index 1 has already been applied.
1142         List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
1143         assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
1144         assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
1145         assertEquals("ApplyJournalEntries index", 1, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
1146
1147         assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
1148         assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
1149         assertEquals("Snapshot getLastAppliedIndex", 1, snapshot.getLastAppliedIndex());
1150         assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
1151         assertEquals("Snapshot getLastIndex", 1, snapshot.getLastIndex());
1152         assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData()),
1153                 MockRaftActor.fromState(snapshot.getState()));
1154     }
1155
1156     @Test
1157     public void testCaptureSnapshotOnMiddleEntryInAppendEntries() {
1158         String id = "testCaptureSnapshotOnMiddleEntryInAppendEntries";
1159         logStart(id);
1160
1161         InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1162
1163         DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1164         config.setSnapshotBatchCount(2);
1165         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1166
1167         final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
1168         RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
1169         Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
1170                 .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
1171         TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
1172                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1173         followerRaftActor.set(followerActorRef.underlyingActor());
1174         followerRaftActor.get().waitForInitializeBehaviorComplete();
1175
1176         InMemorySnapshotStore.addSnapshotSavedLatch(id);
1177         InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1178         InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
1179
1180         List<ReplicatedLogEntry> entries = Arrays.asList(
1181                 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
1182                 newReplicatedLogEntry(1, 2, "three"));
1183
1184         AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 2, -1, (short)0);
1185
1186         followerActorRef.tell(appendEntries, leaderActor);
1187
1188         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1189         assertEquals("isSuccess", true, reply.isSuccess());
1190
1191         final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
1192
1193         InMemoryJournal.waitForDeleteMessagesComplete(id);
1194         InMemoryJournal.waitForWriteMessagesComplete(id);
1195         // We expect the ApplyJournalEntries for index 2 to remain in the persisted log b/c it's still queued for
1196         // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
1197         // This is OK - on recovery it will be a no-op since index 2 has already been applied.
1198         List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
1199         assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
1200         assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
1201         assertEquals("ApplyJournalEntries index", 2, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
1202
1203         assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
1204         assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
1205         assertEquals("Snapshot getLastAppliedIndex", 2, snapshot.getLastAppliedIndex());
1206         assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
1207         assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
1208         assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
1209                 entries.get(2).getData()), MockRaftActor.fromState(snapshot.getState()));
1210
1211         assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
1212         assertEquals("Snapshot index", 2, followerRaftActor.get().getReplicatedLog().getSnapshotIndex());
1213
1214         // Reinstate the actor from persistence
1215
1216         actorFactory.killActor(followerActorRef, new JavaTestKit(getSystem()));
1217
1218         followerActorRef = actorFactory.createTestActor(builder.props()
1219                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1220         followerRaftActor.set(followerActorRef.underlyingActor());
1221         followerRaftActor.get().waitForInitializeBehaviorComplete();
1222
1223         assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
1224         assertEquals("Last index", 2, followerRaftActor.get().getReplicatedLog().lastIndex());
1225         assertEquals("Last applied index", 2, followerRaftActor.get().getRaftActorContext().getLastApplied());
1226         assertEquals("Commit index", 2, followerRaftActor.get().getRaftActorContext().getCommitIndex());
1227         assertEquals("State", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
1228                 entries.get(2).getData()), followerRaftActor.get().getState());
1229     }
1230
1231     @Test
1232     public void testCaptureSnapshotOnAppendEntriesWithUnapplied() {
1233         String id = "testCaptureSnapshotOnAppendEntriesWithUnapplied";
1234         logStart(id);
1235
1236         InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1237
1238         DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1239         config.setSnapshotBatchCount(1);
1240         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1241
1242         final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
1243         RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
1244         Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
1245                 .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
1246         TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
1247                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1248         followerRaftActor.set(followerActorRef.underlyingActor());
1249         followerRaftActor.get().waitForInitializeBehaviorComplete();
1250
1251         InMemorySnapshotStore.addSnapshotSavedLatch(id);
1252         InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1253         InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
1254
1255         List<ReplicatedLogEntry> entries = Arrays.asList(
1256                 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
1257                 newReplicatedLogEntry(1, 2, "three"));
1258
1259         AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 0, -1, (short)0);
1260
1261         followerActorRef.tell(appendEntries, leaderActor);
1262
1263         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1264         assertEquals("isSuccess", true, reply.isSuccess());
1265
1266         final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
1267
1268         InMemoryJournal.waitForDeleteMessagesComplete(id);
1269         InMemoryJournal.waitForWriteMessagesComplete(id);
1270         // We expect the ApplyJournalEntries for index 0 to remain in the persisted log b/c it's still queued for
1271         // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
1272         // This is OK - on recovery it will be a no-op since index 0 has already been applied.
1273         List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
1274         assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
1275         assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
1276         assertEquals("ApplyJournalEntries index", 0, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
1277
1278         assertEquals("Snapshot unapplied size", 2, snapshot.getUnAppliedEntries().size());
1279         assertEquals("Snapshot unapplied entry index", 1, snapshot.getUnAppliedEntries().get(0).getIndex());
1280         assertEquals("Snapshot unapplied entry index", 2, snapshot.getUnAppliedEntries().get(1).getIndex());
1281         assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
1282         assertEquals("Snapshot getLastAppliedIndex", 0, snapshot.getLastAppliedIndex());
1283         assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
1284         assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
1285         assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData()),
1286                 MockRaftActor.fromState(snapshot.getState()));
1287     }
1288
1289     @SuppressWarnings("checkstyle:IllegalCatch")
1290     private static RaftActorSnapshotCohort newRaftActorSnapshotCohort(
1291             final AtomicReference<MockRaftActor> followerRaftActor) {
1292         RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() {
1293             @Override
1294             public void createSnapshot(final ActorRef actorRef,
1295                     final java.util.Optional<OutputStream> installSnapshotStream) {
1296                 try {
1297                     actorRef.tell(new CaptureSnapshotReply(new MockSnapshotState(followerRaftActor.get().getState()),
1298                             installSnapshotStream), actorRef);
1299                 } catch (RuntimeException e) {
1300                     throw e;
1301                 } catch (Exception e) {
1302                     throw new RuntimeException(e);
1303                 }
1304             }
1305
1306             @Override
1307             public void applySnapshot(final State snapshotState) {
1308             }
1309
1310             @Override
1311             public State deserializeSnapshot(final ByteSource snapshotBytes) {
1312                 throw new UnsupportedOperationException();
1313             }
1314         };
1315         return snapshotCohort;
1316     }
1317
1318     public byte[] getNextChunk(final ByteString bs, final int offset, final int chunkSize) {
1319         int snapshotLength = bs.size();
1320         int start = offset;
1321         int size = chunkSize;
1322         if (chunkSize > snapshotLength) {
1323             size = snapshotLength;
1324         } else {
1325             if (start + chunkSize > snapshotLength) {
1326                 size = snapshotLength - start;
1327             }
1328         }
1329
1330         byte[] nextChunk = new byte[size];
1331         bs.copyTo(nextChunk, start, 0, size);
1332         return nextChunk;
1333     }
1334
1335     private void expectAndVerifyAppendEntriesReply(final int expTerm, final boolean expSuccess,
1336             final String expFollowerId, final long expLogLastTerm, final long expLogLastIndex) {
1337         expectAndVerifyAppendEntriesReply(expTerm, expSuccess, expFollowerId, expLogLastTerm, expLogLastIndex, false);
1338     }
1339
1340     private void expectAndVerifyAppendEntriesReply(final int expTerm, final boolean expSuccess,
1341             final String expFollowerId, final long expLogLastTerm, final long expLogLastIndex,
1342             final boolean expForceInstallSnapshot) {
1343
1344         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
1345                 AppendEntriesReply.class);
1346
1347         assertEquals("isSuccess", expSuccess, reply.isSuccess());
1348         assertEquals("getTerm", expTerm, reply.getTerm());
1349         assertEquals("getFollowerId", expFollowerId, reply.getFollowerId());
1350         assertEquals("getLogLastTerm", expLogLastTerm, reply.getLogLastTerm());
1351         assertEquals("getLogLastIndex", expLogLastIndex, reply.getLogLastIndex());
1352         assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion());
1353         assertEquals("isForceInstallSnapshot", expForceInstallSnapshot, reply.isForceInstallSnapshot());
1354     }
1355
1356
1357     private static ReplicatedLogEntry newReplicatedLogEntry(final long term, final long index, final String data) {
1358         return new SimpleReplicatedLogEntry(index, term,
1359                 new MockRaftActorContext.MockPayload(data));
1360     }
1361
1362     private ByteString createSnapshot() {
1363         HashMap<String, String> followerSnapshot = new HashMap<>();
1364         followerSnapshot.put("1", "A");
1365         followerSnapshot.put("2", "B");
1366         followerSnapshot.put("3", "C");
1367
1368         return toByteString(followerSnapshot);
1369     }
1370
1371     @Override
1372     protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(final MockRaftActorContext actorContext,
1373             final ActorRef actorRef, final RaftRPC rpc) {
1374         super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
1375
1376         String expVotedFor = rpc instanceof RequestVote ? ((RequestVote)rpc).getCandidateId() : null;
1377         assertEquals("New votedFor", expVotedFor, actorContext.getTermInformation().getVotedFor());
1378     }
1379
1380     @Override
1381     protected void handleAppendEntriesAddSameEntryToLogReply(final TestActorRef<MessageCollectorActor> replyActor) {
1382         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(replyActor, AppendEntriesReply.class);
1383         assertEquals("isSuccess", true, reply.isSuccess());
1384     }
1385 }

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