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

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