BUG 2773 : Transition Shard to Leader state when it has no peers
[controller.git] / opendaylight / md-sal / sal-akka-raft / src / test / java / org / opendaylight / controller / cluster / raft / behaviors / FollowerTest.java
1 package org.opendaylight.controller.cluster.raft.behaviors;
2
3 import static org.junit.Assert.assertEquals;
4 import static org.junit.Assert.assertFalse;
5 import static org.junit.Assert.assertNotNull;
6 import static org.junit.Assert.assertNull;
7 import static org.junit.Assert.assertTrue;
8 import static org.mockito.Mockito.doReturn;
9 import static org.mockito.Mockito.mock;
10 import static org.mockito.Mockito.never;
11 import static org.mockito.Mockito.verify;
12 import akka.actor.ActorRef;
13 import akka.actor.Props;
14 import akka.testkit.TestActorRef;
15 import com.google.common.base.Stopwatch;
16 import com.google.protobuf.ByteString;
17 import java.util.ArrayList;
18 import java.util.Arrays;
19 import java.util.HashMap;
20 import java.util.List;
21 import java.util.concurrent.TimeUnit;
22 import org.junit.After;
23 import org.junit.Assert;
24 import org.junit.Test;
25 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
26 import org.opendaylight.controller.cluster.raft.RaftActorContext;
27 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
28 import org.opendaylight.controller.cluster.raft.Snapshot;
29 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
30 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
31 import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
32 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
33 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
34 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
35 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
36 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
37 import org.opendaylight.controller.cluster.raft.messages.RequestVote;
38 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
39 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
40
41 public class FollowerTest extends AbstractRaftActorBehaviorTest {
42
43     private final TestActorRef<MessageCollectorActor> followerActor = actorFactory.createTestActor(
44             Props.create(MessageCollectorActor.class), actorFactory.generateActorId("follower"));
45
46     private final TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
47             Props.create(MessageCollectorActor.class), actorFactory.generateActorId("leader"));
48
49     private RaftActorBehavior follower;
50
51     @Override
52     @After
53     public void tearDown() throws Exception {
54         if(follower != null) {
55             follower.close();
56         }
57
58         super.tearDown();
59     }
60
61     @Override
62     protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
63         return new Follower(actorContext);
64     }
65
66     @Override
67     protected  MockRaftActorContext createActorContext() {
68         return createActorContext(followerActor);
69     }
70
71     @Override
72     protected  MockRaftActorContext createActorContext(ActorRef actorRef){
73         return new MockRaftActorContext("follower", getSystem(), actorRef);
74     }
75
76     @Test
77     public void testThatAnElectionTimeoutIsTriggered(){
78         MockRaftActorContext actorContext = createActorContext();
79         follower = new Follower(actorContext);
80
81         MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class,
82                 actorContext.getConfigParams().getElectionTimeOutInterval().$times(6).toMillis());
83     }
84
85     @Test
86     public void testHandleElectionTimeout(){
87         logStart("testHandleElectionTimeout");
88
89         follower = new Follower(createActorContext());
90
91         RaftActorBehavior raftBehavior = follower.handleMessage(followerActor, new ElectionTimeout());
92
93         assertTrue(raftBehavior instanceof Candidate);
94     }
95
96     @Test
97     public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull(){
98         logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull");
99
100         RaftActorContext context = createActorContext();
101         long term = 1000;
102         context.getTermInformation().update(term, null);
103
104         follower = createBehavior(context);
105
106         follower.handleMessage(leaderActor, new RequestVote(term, "test", 10000, 999));
107
108         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
109
110         assertEquals("isVoteGranted", true, reply.isVoteGranted());
111         assertEquals("getTerm", term, reply.getTerm());
112     }
113
114     @Test
115     public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId(){
116         logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId");
117
118         RaftActorContext context = createActorContext();
119         long term = 1000;
120         context.getTermInformation().update(term, "test");
121
122         follower = createBehavior(context);
123
124         follower.handleMessage(leaderActor, new RequestVote(term, "candidate", 10000, 999));
125
126         RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
127
128         assertEquals("isVoteGranted", false, reply.isVoteGranted());
129     }
130
131
132     @Test
133     public void testHandleFirstAppendEntries() throws Exception {
134         logStart("testHandleFirstAppendEntries");
135
136         MockRaftActorContext context = createActorContext();
137
138         List<ReplicatedLogEntry> entries = Arrays.asList(
139                 newReplicatedLogEntry(2, 101, "foo"));
140
141         // The new commitIndex is 101
142         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
143
144         follower = createBehavior(context);
145         follower.handleMessage(leaderActor, appendEntries);
146
147         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
148
149         assertFalse(syncStatus.isInitialSyncDone());
150     }
151
152     @Test
153     public void testHandleSyncUpAppendEntries() throws Exception {
154         logStart("testHandleSyncUpAppendEntries");
155
156         MockRaftActorContext context = createActorContext();
157
158         List<ReplicatedLogEntry> entries = Arrays.asList(
159                 newReplicatedLogEntry(2, 101, "foo"));
160
161         // The new commitIndex is 101
162         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
163
164         follower = createBehavior(context);
165         follower.handleMessage(leaderActor, appendEntries);
166
167         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
168
169         assertFalse(syncStatus.isInitialSyncDone());
170
171         // Clear all the messages
172         followerActor.underlyingActor().clear();
173
174         context.setLastApplied(101);
175         context.setCommitIndex(101);
176         setLastLogEntry(context, 1, 101,
177                 new MockRaftActorContext.MockPayload(""));
178
179         entries = Arrays.asList(
180                 newReplicatedLogEntry(2, 101, "foo"));
181
182         // The new commitIndex is 101
183         appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101);
184         follower.handleMessage(leaderActor, appendEntries);
185
186         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
187
188         assertTrue(syncStatus.isInitialSyncDone());
189
190         followerActor.underlyingActor().clear();
191
192         // Sending the same message again should not generate another message
193
194         follower.handleMessage(leaderActor, appendEntries);
195
196         syncStatus = MessageCollectorActor.getFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
197
198         assertNull(syncStatus);
199
200     }
201
202     @Test
203     public void testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete() throws Exception {
204         logStart("testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete");
205
206         MockRaftActorContext context = createActorContext();
207
208         List<ReplicatedLogEntry> entries = Arrays.asList(
209                 newReplicatedLogEntry(2, 101, "foo"));
210
211         // The new commitIndex is 101
212         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
213
214         follower = createBehavior(context);
215         follower.handleMessage(leaderActor, appendEntries);
216
217         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
218
219         assertFalse(syncStatus.isInitialSyncDone());
220
221         // Clear all the messages
222         followerActor.underlyingActor().clear();
223
224         context.setLastApplied(100);
225         setLastLogEntry(context, 1, 100,
226                 new MockRaftActorContext.MockPayload(""));
227
228         entries = Arrays.asList(
229                 newReplicatedLogEntry(2, 101, "foo"));
230
231         // leader-2 is becoming the leader now and it says the commitIndex is 45
232         appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100);
233         follower.handleMessage(leaderActor, appendEntries);
234
235         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
236
237         // We get a new message saying initial status is not done
238         assertFalse(syncStatus.isInitialSyncDone());
239
240     }
241
242
243     @Test
244     public void testHandleAppendEntriesLeaderChangedAfterSyncUpComplete() throws Exception {
245         logStart("testHandleAppendEntriesLeaderChangedAfterSyncUpComplete");
246
247         MockRaftActorContext context = createActorContext();
248
249         List<ReplicatedLogEntry> entries = Arrays.asList(
250                 newReplicatedLogEntry(2, 101, "foo"));
251
252         // The new commitIndex is 101
253         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
254
255         follower = createBehavior(context);
256         follower.handleMessage(leaderActor, appendEntries);
257
258         FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
259
260         assertFalse(syncStatus.isInitialSyncDone());
261
262         // Clear all the messages
263         followerActor.underlyingActor().clear();
264
265         context.setLastApplied(101);
266         context.setCommitIndex(101);
267         setLastLogEntry(context, 1, 101,
268                 new MockRaftActorContext.MockPayload(""));
269
270         entries = Arrays.asList(
271                 newReplicatedLogEntry(2, 101, "foo"));
272
273         // The new commitIndex is 101
274         appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101);
275         follower.handleMessage(leaderActor, appendEntries);
276
277         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
278
279         assertTrue(syncStatus.isInitialSyncDone());
280
281         // Clear all the messages
282         followerActor.underlyingActor().clear();
283
284         context.setLastApplied(100);
285         setLastLogEntry(context, 1, 100,
286                 new MockRaftActorContext.MockPayload(""));
287
288         entries = Arrays.asList(
289                 newReplicatedLogEntry(2, 101, "foo"));
290
291         // leader-2 is becoming the leader now and it says the commitIndex is 45
292         appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100);
293         follower.handleMessage(leaderActor, appendEntries);
294
295         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
296
297         // We get a new message saying initial status is not done
298         assertFalse(syncStatus.isInitialSyncDone());
299
300     }
301
302
303     /**
304      * This test verifies that when an AppendEntries RPC is received by a RaftActor
305      * with a commitIndex that is greater than what has been applied to the
306      * state machine of the RaftActor, the RaftActor applies the state and
307      * sets it current applied state to the commitIndex of the sender.
308      *
309      * @throws Exception
310      */
311     @Test
312     public void testHandleAppendEntriesWithNewerCommitIndex() throws Exception {
313         logStart("testHandleAppendEntriesWithNewerCommitIndex");
314
315         MockRaftActorContext context = createActorContext();
316
317         context.setLastApplied(100);
318         setLastLogEntry(context, 1, 100,
319                 new MockRaftActorContext.MockPayload(""));
320         context.getReplicatedLog().setSnapshotIndex(99);
321
322         List<ReplicatedLogEntry> entries = Arrays.<ReplicatedLogEntry>asList(
323                 newReplicatedLogEntry(2, 101, "foo"));
324
325         // The new commitIndex is 101
326         AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100);
327
328         follower = createBehavior(context);
329         follower.handleMessage(leaderActor, appendEntries);
330
331         assertEquals("getLastApplied", 101L, context.getLastApplied());
332     }
333
334     /**
335      * This test verifies that when an AppendEntries is received a specific prevLogTerm
336      * which does not match the term that is in RaftActors log entry at prevLogIndex
337      * then the RaftActor does not change it's state and it returns a failure.
338      *
339      * @throws Exception
340      */
341     @Test
342     public void testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() {
343         logStart("testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm");
344
345         MockRaftActorContext context = createActorContext();
346
347         // First set the receivers term to lower number
348         context.getTermInformation().update(95, "test");
349
350         // AppendEntries is now sent with a bigger term
351         // this will set the receivers term to be the same as the sender's term
352         AppendEntries appendEntries = new AppendEntries(100, "leader", 0, 0, null, 101, -1);
353
354         follower = createBehavior(context);
355
356         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
357
358         Assert.assertSame(follower, newBehavior);
359
360         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
361                 AppendEntriesReply.class);
362
363         assertEquals("isSuccess", false, reply.isSuccess());
364     }
365
366     /**
367      * This test verifies that when a new AppendEntries message is received with
368      * new entries and the logs of the sender and receiver match that the new
369      * entries get added to the log and the log is incremented by the number of
370      * entries received in appendEntries
371      *
372      * @throws Exception
373      */
374     @Test
375     public void testHandleAppendEntriesAddNewEntries() {
376         logStart("testHandleAppendEntriesAddNewEntries");
377
378         MockRaftActorContext context = createActorContext();
379
380         // First set the receivers term to lower number
381         context.getTermInformation().update(1, "test");
382
383         // Prepare the receivers log
384         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
385         log.append(newReplicatedLogEntry(1, 0, "zero"));
386         log.append(newReplicatedLogEntry(1, 1, "one"));
387         log.append(newReplicatedLogEntry(1, 2, "two"));
388
389         context.setReplicatedLog(log);
390
391         // Prepare the entries to be sent with AppendEntries
392         List<ReplicatedLogEntry> entries = new ArrayList<>();
393         entries.add(newReplicatedLogEntry(1, 3, "three"));
394         entries.add(newReplicatedLogEntry(1, 4, "four"));
395
396         // Send appendEntries with the same term as was set on the receiver
397         // before the new behavior was created (1 in this case)
398         // This will not work for a Candidate because as soon as a Candidate
399         // is created it increments the term
400         AppendEntries appendEntries = new AppendEntries(1, "leader-1", 2, 1, entries, 4, -1);
401
402         follower = createBehavior(context);
403
404         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
405
406         Assert.assertSame(follower, newBehavior);
407
408         assertEquals("Next index", 5, log.last().getIndex() + 1);
409         assertEquals("Entry 3", entries.get(0), log.get(3));
410         assertEquals("Entry 4", entries.get(1), log.get(4));
411
412         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
413     }
414
415     /**
416      * This test verifies that when a new AppendEntries message is received with
417      * new entries and the logs of the sender and receiver are out-of-sync that
418      * the log is first corrected by removing the out of sync entries from the
419      * log and then adding in the new entries sent with the AppendEntries message
420      */
421     @Test
422     public void testHandleAppendEntriesCorrectReceiverLogEntries() {
423         logStart("testHandleAppendEntriesCorrectReceiverLogEntries");
424
425         MockRaftActorContext context = createActorContext();
426
427         // First set the receivers term to lower number
428         context.getTermInformation().update(1, "test");
429
430         // Prepare the receivers log
431         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
432         log.append(newReplicatedLogEntry(1, 0, "zero"));
433         log.append(newReplicatedLogEntry(1, 1, "one"));
434         log.append(newReplicatedLogEntry(1, 2, "two"));
435
436         context.setReplicatedLog(log);
437
438         // Prepare the entries to be sent with AppendEntries
439         List<ReplicatedLogEntry> entries = new ArrayList<>();
440         entries.add(newReplicatedLogEntry(2, 2, "two-1"));
441         entries.add(newReplicatedLogEntry(2, 3, "three"));
442
443         // Send appendEntries with the same term as was set on the receiver
444         // before the new behavior was created (1 in this case)
445         // This will not work for a Candidate because as soon as a Candidate
446         // is created it increments the term
447         AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1);
448
449         follower = createBehavior(context);
450
451         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
452
453         Assert.assertSame(follower, newBehavior);
454
455         // The entry at index 2 will be found out-of-sync with the leader
456         // and will be removed
457         // Then the two new entries will be added to the log
458         // Thus making the log to have 4 entries
459         assertEquals("Next index", 4, log.last().getIndex() + 1);
460         //assertEquals("Entry 2", entries.get(0), log.get(2));
461
462         assertEquals("Entry 1 data", "one", log.get(1).getData().toString());
463
464         // Check that the entry at index 2 has the new data
465         assertEquals("Entry 2", entries.get(0), log.get(2));
466
467         assertEquals("Entry 3", entries.get(1), log.get(3));
468
469         expectAndVerifyAppendEntriesReply(2, true, context.getId(), 2, 3);
470     }
471
472     @Test
473     public void testHandleAppendEntriesPreviousLogEntryMissing(){
474         logStart("testHandleAppendEntriesPreviousLogEntryMissing");
475
476         MockRaftActorContext context = createActorContext();
477
478         // Prepare the receivers log
479         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
480         log.append(newReplicatedLogEntry(1, 0, "zero"));
481         log.append(newReplicatedLogEntry(1, 1, "one"));
482         log.append(newReplicatedLogEntry(1, 2, "two"));
483
484         context.setReplicatedLog(log);
485
486         // Prepare the entries to be sent with AppendEntries
487         List<ReplicatedLogEntry> entries = new ArrayList<>();
488         entries.add(newReplicatedLogEntry(1, 4, "four"));
489
490         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1);
491
492         follower = createBehavior(context);
493
494         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
495
496         Assert.assertSame(follower, newBehavior);
497
498         expectAndVerifyAppendEntriesReply(1, false, context.getId(), 1, 2);
499     }
500
501     @Test
502     public void testHandleAppendEntriesWithExistingLogEntry() {
503         logStart("testHandleAppendEntriesWithExistingLogEntry");
504
505         MockRaftActorContext context = createActorContext();
506
507         context.getTermInformation().update(1, "test");
508
509         // Prepare the receivers log
510         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
511         log.append(newReplicatedLogEntry(1, 0, "zero"));
512         log.append(newReplicatedLogEntry(1, 1, "one"));
513
514         context.setReplicatedLog(log);
515
516         // Send the last entry again.
517         List<ReplicatedLogEntry> entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"));
518
519         follower = createBehavior(context);
520
521         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1));
522
523         assertEquals("Next index", 2, log.last().getIndex() + 1);
524         assertEquals("Entry 1", entries.get(0), log.get(1));
525
526         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 1);
527
528         // Send the last entry again and also a new one.
529
530         entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two"));
531
532         leaderActor.underlyingActor().clear();
533         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1));
534
535         assertEquals("Next index", 3, log.last().getIndex() + 1);
536         assertEquals("Entry 1", entries.get(0), log.get(1));
537         assertEquals("Entry 2", entries.get(1), log.get(2));
538
539         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 2);
540     }
541
542     @Test
543     public void testHandleAppendEntriesAfterInstallingSnapshot(){
544         logStart("testHandleAppendAfterInstallingSnapshot");
545
546         MockRaftActorContext context = createActorContext();
547
548         // Prepare the receivers log
549         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
550
551         // Set up a log as if it has been snapshotted
552         log.setSnapshotIndex(3);
553         log.setSnapshotTerm(1);
554
555         context.setReplicatedLog(log);
556
557         // Prepare the entries to be sent with AppendEntries
558         List<ReplicatedLogEntry> entries = new ArrayList<>();
559         entries.add(newReplicatedLogEntry(1, 4, "four"));
560
561         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3);
562
563         follower = createBehavior(context);
564
565         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
566
567         Assert.assertSame(follower, newBehavior);
568
569         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
570     }
571
572
573     /**
574      * This test verifies that when InstallSnapshot is received by
575      * the follower its applied correctly.
576      *
577      * @throws Exception
578      */
579     @Test
580     public void testHandleInstallSnapshot() throws Exception {
581         logStart("testHandleInstallSnapshot");
582
583         MockRaftActorContext context = createActorContext();
584
585         follower = createBehavior(context);
586
587         ByteString bsSnapshot  = createSnapshot();
588         int offset = 0;
589         int snapshotLength = bsSnapshot.size();
590         int chunkSize = 50;
591         int totalChunks = (snapshotLength / chunkSize) + ((snapshotLength % chunkSize) > 0 ? 1 : 0);
592         int lastIncludedIndex = 1;
593         int chunkIndex = 1;
594         InstallSnapshot lastInstallSnapshot = null;
595
596         for(int i = 0; i < totalChunks; i++) {
597             ByteString chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
598             lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
599                     chunkData, chunkIndex, totalChunks);
600             follower.handleMessage(leaderActor, lastInstallSnapshot);
601             offset = offset + 50;
602             lastIncludedIndex++;
603             chunkIndex++;
604         }
605
606         ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
607                 ApplySnapshot.class);
608         Snapshot snapshot = applySnapshot.getSnapshot();
609         assertEquals("getLastIndex", lastInstallSnapshot.getLastIncludedIndex(), snapshot.getLastIndex());
610         assertEquals("getLastIncludedTerm", lastInstallSnapshot.getLastIncludedTerm(),
611                 snapshot.getLastAppliedTerm());
612         assertEquals("getLastAppliedIndex", lastInstallSnapshot.getLastIncludedIndex(),
613                 snapshot.getLastAppliedIndex());
614         assertEquals("getLastTerm", lastInstallSnapshot.getLastIncludedTerm(), snapshot.getLastTerm());
615         Assert.assertArrayEquals("getState", bsSnapshot.toByteArray(), snapshot.getState());
616
617         List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
618                 leaderActor, InstallSnapshotReply.class);
619         assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
620
621         chunkIndex = 1;
622         for(InstallSnapshotReply reply: replies) {
623             assertEquals("getChunkIndex", chunkIndex++, reply.getChunkIndex());
624             assertEquals("getTerm", 1, reply.getTerm());
625             assertEquals("isSuccess", true, reply.isSuccess());
626             assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
627         }
628
629         assertNull("Expected null SnapshotTracker", ((Follower) follower).getSnapshotTracker());
630     }
631
632
633     /**
634      * Verify that when an AppendEntries is sent to a follower during a snapshot install
635      * the Follower short-circuits the processing of the AppendEntries message.
636      *
637      * @throws Exception
638      */
639     @Test
640     public void testReceivingAppendEntriesDuringInstallSnapshot() throws Exception {
641         logStart("testReceivingAppendEntriesDuringInstallSnapshot");
642
643         MockRaftActorContext context = createActorContext();
644
645         follower = createBehavior(context);
646
647         ByteString bsSnapshot  = createSnapshot();
648         int snapshotLength = bsSnapshot.size();
649         int chunkSize = 50;
650         int totalChunks = (snapshotLength / chunkSize) + ((snapshotLength % chunkSize) > 0 ? 1 : 0);
651         int lastIncludedIndex = 1;
652
653         // Check that snapshot installation is not in progress
654         assertNull(((Follower) follower).getSnapshotTracker());
655
656         // Make sure that we have more than 1 chunk to send
657         assertTrue(totalChunks > 1);
658
659         // Send an install snapshot with the first chunk to start the process of installing a snapshot
660         ByteString chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
661         follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
662                 chunkData, 1, totalChunks));
663
664         // Check if snapshot installation is in progress now
665         assertNotNull(((Follower) follower).getSnapshotTracker());
666
667         // Send an append entry
668         AppendEntries appendEntries = mock(AppendEntries.class);
669         doReturn(context.getTermInformation().getCurrentTerm()).when(appendEntries).getTerm();
670
671         follower.handleMessage(leaderActor, appendEntries);
672
673         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
674         assertEquals(context.getReplicatedLog().lastIndex(), reply.getLogLastIndex());
675         assertEquals(context.getReplicatedLog().lastTerm(), reply.getLogLastTerm());
676         assertEquals(context.getTermInformation().getCurrentTerm(), reply.getTerm());
677
678         // We should not hit the code that needs to look at prevLogIndex because we are short circuiting
679         verify(appendEntries, never()).getPrevLogIndex();
680
681     }
682
683     @Test
684     public void testInitialSyncUpWithHandleInstallSnapshotFollowedByAppendEntries() throws Exception {
685         logStart("testInitialSyncUpWithHandleInstallSnapshot");
686
687         MockRaftActorContext context = createActorContext();
688
689         follower = createBehavior(context);
690
691         ByteString bsSnapshot  = createSnapshot();
692         int offset = 0;
693         int snapshotLength = bsSnapshot.size();
694         int chunkSize = 50;
695         int totalChunks = (snapshotLength / chunkSize) + ((snapshotLength % chunkSize) > 0 ? 1 : 0);
696         int lastIncludedIndex = 1;
697         int chunkIndex = 1;
698         InstallSnapshot lastInstallSnapshot = null;
699
700         for(int i = 0; i < totalChunks; i++) {
701             ByteString chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
702             lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
703                     chunkData, chunkIndex, totalChunks);
704             follower.handleMessage(leaderActor, lastInstallSnapshot);
705             offset = offset + 50;
706             lastIncludedIndex++;
707             chunkIndex++;
708         }
709
710         FollowerInitialSyncUpStatus syncStatus =
711                 MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
712
713         assertFalse(syncStatus.isInitialSyncDone());
714
715         // Clear all the messages
716         followerActor.underlyingActor().clear();
717
718         context.setLastApplied(101);
719         context.setCommitIndex(101);
720         setLastLogEntry(context, 1, 101,
721                 new MockRaftActorContext.MockPayload(""));
722
723         List<ReplicatedLogEntry> entries = Arrays.asList(
724                 newReplicatedLogEntry(2, 101, "foo"));
725
726         // The new commitIndex is 101
727         AppendEntries appendEntries = new AppendEntries(2, "leader", 101, 1, entries, 102, 101);
728         follower.handleMessage(leaderActor, appendEntries);
729
730         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
731
732         assertTrue(syncStatus.isInitialSyncDone());
733     }
734
735     @Test
736     public void testHandleOutOfSequenceInstallSnapshot() throws Exception {
737         logStart("testHandleOutOfSequenceInstallSnapshot");
738
739         MockRaftActorContext context = createActorContext();
740
741         follower = createBehavior(context);
742
743         ByteString bsSnapshot = createSnapshot();
744
745         InstallSnapshot installSnapshot = new InstallSnapshot(1, "leader", 3, 1,
746                 getNextChunk(bsSnapshot, 10, 50), 3, 3);
747         follower.handleMessage(leaderActor, installSnapshot);
748
749         InstallSnapshotReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
750                 InstallSnapshotReply.class);
751
752         assertEquals("isSuccess", false, reply.isSuccess());
753         assertEquals("getChunkIndex", -1, reply.getChunkIndex());
754         assertEquals("getTerm", 1, reply.getTerm());
755         assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
756
757         assertNull("Expected null SnapshotTracker", ((Follower) follower).getSnapshotTracker());
758     }
759
760     @Test
761     public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers(){
762         MockRaftActorContext context = createActorContext();
763
764         Stopwatch stopwatch = Stopwatch.createStarted();
765
766         follower = createBehavior(context);
767
768         MessageCollectorActor.expectFirstMatching(followerActor, ElectionTimeout.class);
769
770         long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
771
772         assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis());
773     }
774
775     public ByteString getNextChunk (ByteString bs, int offset, int chunkSize){
776         int snapshotLength = bs.size();
777         int start = offset;
778         int size = chunkSize;
779         if (chunkSize > snapshotLength) {
780             size = snapshotLength;
781         } else {
782             if ((start + chunkSize) > snapshotLength) {
783                 size = snapshotLength - start;
784             }
785         }
786         return bs.substring(start, start + size);
787     }
788
789     private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
790             String expFollowerId, long expLogLastTerm, long expLogLastIndex) {
791
792         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
793                 AppendEntriesReply.class);
794
795         assertEquals("isSuccess", expSuccess, reply.isSuccess());
796         assertEquals("getTerm", expTerm, reply.getTerm());
797         assertEquals("getFollowerId", expFollowerId, reply.getFollowerId());
798         assertEquals("getLogLastTerm", expLogLastTerm, reply.getLogLastTerm());
799         assertEquals("getLogLastIndex", expLogLastIndex, reply.getLogLastIndex());
800     }
801
802     private ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
803         return new MockRaftActorContext.MockReplicatedLogEntry(term, index,
804                 new MockRaftActorContext.MockPayload(data));
805     }
806
807     private ByteString createSnapshot(){
808         HashMap<String, String> followerSnapshot = new HashMap<>();
809         followerSnapshot.put("1", "A");
810         followerSnapshot.put("2", "B");
811         followerSnapshot.put("3", "C");
812
813         return toByteString(followerSnapshot);
814     }
815
816     @Override
817     protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(RaftActorContext actorContext,
818             ActorRef actorRef, RaftRPC rpc) throws Exception {
819         super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
820
821         String expVotedFor = RequestVote.class.isInstance(rpc) ? ((RequestVote)rpc).getCandidateId() : null;
822         assertEquals("New votedFor", expVotedFor, actorContext.getTermInformation().getVotedFor());
823     }
824
825     @Override
826     protected void handleAppendEntriesAddSameEntryToLogReply(TestActorRef<MessageCollectorActor> replyActor)
827             throws Exception {
828         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(replyActor, AppendEntriesReply.class);
829         assertEquals("isSuccess", true, reply.isSuccess());
830     }
831 }

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