2 * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.controller.cluster.raft.behaviors;
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;
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;
80 public class FollowerTest extends AbstractRaftActorBehaviorTest<Follower> {
82 private final TestActorRef<MessageCollectorActor> followerActor = actorFactory.createTestActor(
83 Props.create(MessageCollectorActor.class), actorFactory.generateActorId("follower"));
85 private final TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
86 Props.create(MessageCollectorActor.class), actorFactory.generateActorId("leader"));
88 private Follower follower;
90 private final short payloadVersion = 5;
94 public void tearDown() throws Exception {
95 if (follower != null) {
103 protected Follower createBehavior(RaftActorContext actorContext) {
104 return spy(new Follower(actorContext));
108 protected MockRaftActorContext createActorContext() {
109 return createActorContext(followerActor);
113 protected MockRaftActorContext createActorContext(ActorRef actorRef) {
114 MockRaftActorContext context = new MockRaftActorContext("follower", getSystem(), actorRef);
115 context.setPayloadVersion(payloadVersion);
120 public void testThatAnElectionTimeoutIsTriggered() {
121 MockRaftActorContext actorContext = createActorContext();
122 follower = new Follower(actorContext);
124 MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class,
125 actorContext.getConfigParams().getElectionTimeOutInterval().$times(6).toMillis());
129 public void testHandleElectionTimeoutWhenNoLeaderMessageReceived() {
130 logStart("testHandleElectionTimeoutWhenNoLeaderMessageReceived");
132 MockRaftActorContext context = createActorContext();
133 follower = new Follower(context);
135 Uninterruptibles.sleepUninterruptibly(context.getConfigParams().getElectionTimeOutInterval().toMillis(),
136 TimeUnit.MILLISECONDS);
137 RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
139 assertTrue(raftBehavior instanceof Candidate);
143 public void testHandleElectionTimeoutWhenLeaderMessageReceived() {
144 logStart("testHandleElectionTimeoutWhenLeaderMessageReceived");
146 MockRaftActorContext context = createActorContext();
147 ((DefaultConfigParamsImpl) context.getConfigParams())
148 .setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
149 ((DefaultConfigParamsImpl) context.getConfigParams()).setElectionTimeoutFactor(4);
151 follower = new Follower(context);
152 context.setCurrentBehavior(follower);
154 Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
155 .getElectionTimeOutInterval().toMillis() - 100, TimeUnit.MILLISECONDS);
156 follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
159 Uninterruptibles.sleepUninterruptibly(130, TimeUnit.MILLISECONDS);
160 RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
161 assertTrue(raftBehavior instanceof Follower);
163 Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
164 .getElectionTimeOutInterval().toMillis() - 150, TimeUnit.MILLISECONDS);
165 follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
168 Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
169 raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
170 assertTrue(raftBehavior instanceof Follower);
174 public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull() {
175 logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull");
177 MockRaftActorContext context = createActorContext();
179 context.getTermInformation().update(term, null);
181 follower = createBehavior(context);
183 follower.handleMessage(leaderActor, new RequestVote(term, "test", 10000, 999));
185 RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
187 assertEquals("isVoteGranted", true, reply.isVoteGranted());
188 assertEquals("getTerm", term, reply.getTerm());
189 verify(follower).scheduleElection(any(FiniteDuration.class));
193 public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId() {
194 logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId");
196 MockRaftActorContext context = createActorContext();
198 context.getTermInformation().update(term, "test");
200 follower = createBehavior(context);
202 follower.handleMessage(leaderActor, new RequestVote(term, "candidate", 10000, 999));
204 RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
206 assertEquals("isVoteGranted", false, reply.isVoteGranted());
207 verify(follower, never()).scheduleElection(any(FiniteDuration.class));
212 public void testHandleFirstAppendEntries() throws Exception {
213 logStart("testHandleFirstAppendEntries");
215 MockRaftActorContext context = createActorContext();
216 context.getReplicatedLog().clear(0,2);
217 context.getReplicatedLog().append(newReplicatedLogEntry(1,100, "bar"));
218 context.getReplicatedLog().setSnapshotIndex(99);
220 List<ReplicatedLogEntry> entries = Arrays.asList(
221 newReplicatedLogEntry(2, 101, "foo"));
223 Assert.assertEquals(1, context.getReplicatedLog().size());
225 // The new commitIndex is 101
226 AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
228 follower = createBehavior(context);
229 follower.handleMessage(leaderActor, appendEntries);
231 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
232 FollowerInitialSyncUpStatus.class);
233 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
235 assertFalse(syncStatus.isInitialSyncDone());
236 assertTrue("append entries reply should be true", reply.isSuccess());
240 public void testHandleFirstAppendEntriesWithPrevIndexMinusOne() throws Exception {
241 logStart("testHandleFirstAppendEntries");
243 MockRaftActorContext context = createActorContext();
245 List<ReplicatedLogEntry> entries = Arrays.asList(
246 newReplicatedLogEntry(2, 101, "foo"));
248 // The new commitIndex is 101
249 AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
251 follower = createBehavior(context);
252 follower.handleMessage(leaderActor, appendEntries);
254 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
255 FollowerInitialSyncUpStatus.class);
256 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
258 assertFalse(syncStatus.isInitialSyncDone());
259 assertFalse("append entries reply should be false", reply.isSuccess());
263 public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog()
265 logStart("testHandleFirstAppendEntries");
267 MockRaftActorContext context = createActorContext();
268 context.getReplicatedLog().clear(0,2);
269 context.getReplicatedLog().append(newReplicatedLogEntry(1, 100, "bar"));
270 context.getReplicatedLog().setSnapshotIndex(99);
272 List<ReplicatedLogEntry> entries = Arrays.asList(
273 newReplicatedLogEntry(2, 101, "foo"));
275 // The new commitIndex is 101
276 AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
278 follower = createBehavior(context);
279 follower.handleMessage(leaderActor, appendEntries);
281 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
282 FollowerInitialSyncUpStatus.class);
283 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
285 assertFalse(syncStatus.isInitialSyncDone());
286 assertTrue("append entries reply should be true", reply.isSuccess());
290 public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot()
292 logStart("testHandleFirstAppendEntries");
294 MockRaftActorContext context = createActorContext();
295 context.getReplicatedLog().clear(0,2);
296 context.getReplicatedLog().setSnapshotIndex(100);
298 List<ReplicatedLogEntry> entries = Arrays.asList(
299 newReplicatedLogEntry(2, 101, "foo"));
301 // The new commitIndex is 101
302 AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
304 follower = createBehavior(context);
305 follower.handleMessage(leaderActor, appendEntries);
307 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
308 FollowerInitialSyncUpStatus.class);
309 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
311 assertFalse(syncStatus.isInitialSyncDone());
312 assertTrue("append entries reply should be true", reply.isSuccess());
316 public void testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing()
319 "testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing");
321 MockRaftActorContext context = createActorContext();
322 context.getReplicatedLog().clear(0,2);
323 context.getReplicatedLog().setSnapshotIndex(100);
325 List<ReplicatedLogEntry> entries = Arrays.asList(
326 newReplicatedLogEntry(2, 105, "foo"));
328 // The new commitIndex is 101
329 AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 105, 100, (short) 0);
331 follower = createBehavior(context);
332 follower.handleMessage(leaderActor, appendEntries);
334 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
335 FollowerInitialSyncUpStatus.class);
336 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
338 assertFalse(syncStatus.isInitialSyncDone());
339 assertFalse("append entries reply should be false", reply.isSuccess());
343 public void testHandleSyncUpAppendEntries() throws Exception {
344 logStart("testHandleSyncUpAppendEntries");
346 MockRaftActorContext context = createActorContext();
348 List<ReplicatedLogEntry> entries = Arrays.asList(
349 newReplicatedLogEntry(2, 101, "foo"));
351 // The new commitIndex is 101
352 AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
354 follower = createBehavior(context);
355 follower.handleMessage(leaderActor, appendEntries);
357 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
358 FollowerInitialSyncUpStatus.class);
360 assertFalse(syncStatus.isInitialSyncDone());
362 // Clear all the messages
363 followerActor.underlyingActor().clear();
365 context.setLastApplied(101);
366 context.setCommitIndex(101);
367 setLastLogEntry(context, 1, 101,
368 new MockRaftActorContext.MockPayload(""));
370 entries = Arrays.asList(
371 newReplicatedLogEntry(2, 101, "foo"));
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);
377 syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
379 assertTrue(syncStatus.isInitialSyncDone());
381 followerActor.underlyingActor().clear();
383 // Sending the same message again should not generate another message
385 follower.handleMessage(leaderActor, appendEntries);
387 syncStatus = MessageCollectorActor.getFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
389 assertNull(syncStatus);
394 public void testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete() throws Exception {
395 logStart("testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete");
397 MockRaftActorContext context = createActorContext();
399 List<ReplicatedLogEntry> entries = Arrays.asList(
400 newReplicatedLogEntry(2, 101, "foo"));
402 // The new commitIndex is 101
403 AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
405 follower = createBehavior(context);
406 follower.handleMessage(leaderActor, appendEntries);
408 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
409 FollowerInitialSyncUpStatus.class);
411 assertFalse(syncStatus.isInitialSyncDone());
413 // Clear all the messages
414 followerActor.underlyingActor().clear();
416 context.setLastApplied(100);
417 setLastLogEntry(context, 1, 100,
418 new MockRaftActorContext.MockPayload(""));
420 entries = Arrays.asList(
421 newReplicatedLogEntry(2, 101, "foo"));
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);
427 syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
429 // We get a new message saying initial status is not done
430 assertFalse(syncStatus.isInitialSyncDone());
436 public void testHandleAppendEntriesLeaderChangedAfterSyncUpComplete() throws Exception {
437 logStart("testHandleAppendEntriesLeaderChangedAfterSyncUpComplete");
439 MockRaftActorContext context = createActorContext();
441 List<ReplicatedLogEntry> entries = Arrays.asList(
442 newReplicatedLogEntry(2, 101, "foo"));
444 // The new commitIndex is 101
445 AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
447 follower = createBehavior(context);
448 follower.handleMessage(leaderActor, appendEntries);
450 FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
451 FollowerInitialSyncUpStatus.class);
453 assertFalse(syncStatus.isInitialSyncDone());
455 // Clear all the messages
456 followerActor.underlyingActor().clear();
458 context.setLastApplied(101);
459 context.setCommitIndex(101);
460 setLastLogEntry(context, 1, 101,
461 new MockRaftActorContext.MockPayload(""));
463 entries = Arrays.asList(
464 newReplicatedLogEntry(2, 101, "foo"));
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);
470 syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
472 assertTrue(syncStatus.isInitialSyncDone());
474 // Clear all the messages
475 followerActor.underlyingActor().clear();
477 context.setLastApplied(100);
478 setLastLogEntry(context, 1, 100,
479 new MockRaftActorContext.MockPayload(""));
481 entries = Arrays.asList(
482 newReplicatedLogEntry(2, 101, "foo"));
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);
488 syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
490 // We get a new message saying initial status is not done
491 assertFalse(syncStatus.isInitialSyncDone());
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.
503 public void testHandleAppendEntriesWithNewerCommitIndex() throws Exception {
504 logStart("testHandleAppendEntriesWithNewerCommitIndex");
506 MockRaftActorContext context = createActorContext();
508 context.setLastApplied(100);
509 setLastLogEntry(context, 1, 100,
510 new MockRaftActorContext.MockPayload(""));
511 context.getReplicatedLog().setSnapshotIndex(99);
513 List<ReplicatedLogEntry> entries = Arrays.<ReplicatedLogEntry>asList(
514 newReplicatedLogEntry(2, 101, "foo"));
516 // The new commitIndex is 101
517 AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
519 follower = createBehavior(context);
520 follower.handleMessage(leaderActor, appendEntries);
522 assertEquals("getLastApplied", 101L, context.getLastApplied());
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.
531 public void testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() {
532 logStart("testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm");
534 MockRaftActorContext context = createActorContext();
536 // First set the receivers term to lower number
537 context.getTermInformation().update(95, "test");
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,
544 follower = createBehavior(context);
546 RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
548 Assert.assertSame(follower, newBehavior);
550 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
551 AppendEntriesReply.class);
553 assertEquals("isSuccess", false, reply.isSuccess());
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.
563 public void testHandleAppendEntriesAddNewEntries() {
564 logStart("testHandleAppendEntriesAddNewEntries");
566 MockRaftActorContext context = createActorContext();
568 // First set the receivers term to lower number
569 context.getTermInformation().update(1, "test");
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"));
577 context.setReplicatedLog(log);
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"));
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);
592 follower = createBehavior(context);
594 RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
596 Assert.assertSame(follower, newBehavior);
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));
602 assertEquals("getLeaderPayloadVersion", leaderPayloadVersion, newBehavior.getLeaderPayloadVersion());
603 assertEquals("getLeaderId", leaderId, newBehavior.getLeaderId());
605 expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
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.
615 public void testHandleAppendEntriesCorrectReceiverLogEntries() {
616 logStart("testHandleAppendEntriesCorrectReceiverLogEntries");
618 MockRaftActorContext context = createActorContext();
620 // First set the receivers term to lower number
621 context.getTermInformation().update(1, "test");
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"));
629 context.setReplicatedLog(log);
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"));
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);
642 follower = createBehavior(context);
644 RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
646 Assert.assertSame(follower, newBehavior);
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));
655 assertEquals("Entry 1 data", "one", log.get(1).getData().toString());
657 // Check that the entry at index 2 has the new data
658 assertEquals("Entry 2", entries.get(0), log.get(2));
660 assertEquals("Entry 3", entries.get(1), log.get(3));
662 expectAndVerifyAppendEntriesReply(2, true, context.getId(), 2, 3);
666 public void testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot() {
667 logStart("testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot");
669 MockRaftActorContext context = createActorContext();
671 // First set the receivers term to lower number
672 context.getTermInformation().update(1, "test");
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"));
680 context.setReplicatedLog(log);
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"));
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);
693 context.setRaftPolicy(createRaftPolicy(false, true));
694 follower = createBehavior(context);
696 RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
698 Assert.assertSame(follower, newBehavior);
700 expectAndVerifyAppendEntriesReply(2, false, context.getId(), 1, 2, true);
704 public void testHandleAppendEntriesPreviousLogEntryMissing() {
705 logStart("testHandleAppendEntriesPreviousLogEntryMissing");
707 final MockRaftActorContext context = createActorContext();
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"));
715 context.setReplicatedLog(log);
717 // Prepare the entries to be sent with AppendEntries
718 List<ReplicatedLogEntry> entries = new ArrayList<>();
719 entries.add(newReplicatedLogEntry(1, 4, "four"));
721 AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1, (short)0);
723 follower = createBehavior(context);
725 RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
727 Assert.assertSame(follower, newBehavior);
729 expectAndVerifyAppendEntriesReply(1, false, context.getId(), 1, 2);
733 public void testHandleAppendEntriesWithExistingLogEntry() {
734 logStart("testHandleAppendEntriesWithExistingLogEntry");
736 MockRaftActorContext context = createActorContext();
738 context.getTermInformation().update(1, "test");
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"));
745 context.setReplicatedLog(log);
747 // Send the last entry again.
748 List<ReplicatedLogEntry> entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"));
750 follower = createBehavior(context);
752 follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1, (short)0));
754 assertEquals("Next index", 2, log.last().getIndex() + 1);
755 assertEquals("Entry 1", entries.get(0), log.get(1));
757 expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 1);
759 // Send the last entry again and also a new one.
761 entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two"));
763 leaderActor.underlyingActor().clear();
764 follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1, (short)0));
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));
770 expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 2);
774 public void testHandleAppendEntriesAfterInstallingSnapshot() {
775 logStart("testHandleAppendAfterInstallingSnapshot");
777 MockRaftActorContext context = createActorContext();
779 // Prepare the receivers log
780 MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
782 // Set up a log as if it has been snapshotted
783 log.setSnapshotIndex(3);
784 log.setSnapshotTerm(1);
786 context.setReplicatedLog(log);
788 // Prepare the entries to be sent with AppendEntries
789 List<ReplicatedLogEntry> entries = new ArrayList<>();
790 entries.add(newReplicatedLogEntry(1, 4, "four"));
792 AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3, (short)0);
794 follower = createBehavior(context);
796 RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
798 Assert.assertSame(follower, newBehavior);
800 expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
805 * This test verifies that when InstallSnapshot is received by
806 * the follower its applied correctly.
809 public void testHandleInstallSnapshot() throws Exception {
810 logStart("testHandleInstallSnapshot");
812 MockRaftActorContext context = createActorContext();
813 context.getTermInformation().update(1, "leader");
815 follower = createBehavior(context);
817 ByteString bsSnapshot = createSnapshot();
819 int snapshotLength = bsSnapshot.size();
821 int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
822 int lastIncludedIndex = 1;
824 InstallSnapshot lastInstallSnapshot = null;
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;
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();
852 List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
853 leaderActor, InstallSnapshotReply.class);
854 assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
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());
864 assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
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.
873 public void testReceivingAppendEntriesDuringInstallSnapshot() throws Exception {
874 logStart("testReceivingAppendEntriesDuringInstallSnapshot");
876 MockRaftActorContext context = createActorContext();
878 follower = createBehavior(context);
880 ByteString bsSnapshot = createSnapshot();
881 int snapshotLength = bsSnapshot.size();
883 int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
884 int lastIncludedIndex = 1;
886 // Check that snapshot installation is not in progress
887 assertNull(follower.getSnapshotTracker());
889 // Make sure that we have more than 1 chunk to send
890 assertTrue(totalChunks > 1);
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));
897 // Check if snapshot installation is in progress now
898 assertNotNull(follower.getSnapshotTracker());
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);
904 follower.handleMessage(leaderActor, appendEntries);
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());
912 assertNotNull(follower.getSnapshotTracker());
916 public void testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader() throws Exception {
917 logStart("testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader");
919 MockRaftActorContext context = createActorContext();
921 follower = createBehavior(context);
923 ByteString bsSnapshot = createSnapshot();
924 int snapshotLength = bsSnapshot.size();
926 int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
927 int lastIncludedIndex = 1;
929 // Check that snapshot installation is not in progress
930 assertNull(follower.getSnapshotTracker());
932 // Make sure that we have more than 1 chunk to send
933 assertTrue(totalChunks > 1);
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));
940 // Check if snapshot installation is in progress now
941 assertNotNull(follower.getSnapshotTracker());
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);
947 follower.handleMessage(leaderActor, appendEntries);
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());
955 assertNull(follower.getSnapshotTracker());
959 public void testInitialSyncUpWithHandleInstallSnapshotFollowedByAppendEntries() throws Exception {
960 logStart("testInitialSyncUpWithHandleInstallSnapshot");
962 MockRaftActorContext context = createActorContext();
963 context.setCommitIndex(-1);
965 follower = createBehavior(context);
967 ByteString bsSnapshot = createSnapshot();
969 int snapshotLength = bsSnapshot.size();
971 int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
972 int lastIncludedIndex = 1;
974 InstallSnapshot lastInstallSnapshot = null;
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;
986 FollowerInitialSyncUpStatus syncStatus =
987 MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
989 assertFalse(syncStatus.isInitialSyncDone());
991 // Clear all the messages
992 followerActor.underlyingActor().clear();
994 context.setLastApplied(101);
995 context.setCommitIndex(101);
996 setLastLogEntry(context, 1, 101,
997 new MockRaftActorContext.MockPayload(""));
999 List<ReplicatedLogEntry> entries = Arrays.asList(
1000 newReplicatedLogEntry(2, 101, "foo"));
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);
1006 syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
1008 assertTrue(syncStatus.isInitialSyncDone());
1012 public void testHandleOutOfSequenceInstallSnapshot() throws Exception {
1013 logStart("testHandleOutOfSequenceInstallSnapshot");
1015 MockRaftActorContext context = createActorContext();
1017 follower = createBehavior(context);
1019 ByteString bsSnapshot = createSnapshot();
1021 InstallSnapshot installSnapshot = new InstallSnapshot(1, "leader", 3, 1,
1022 getNextChunk(bsSnapshot, 10, 50), 3, 3);
1023 follower.handleMessage(leaderActor, installSnapshot);
1025 InstallSnapshotReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
1026 InstallSnapshotReply.class);
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());
1033 assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
1037 public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers() {
1038 MockRaftActorContext context = createActorContext();
1040 Stopwatch stopwatch = Stopwatch.createStarted();
1042 follower = createBehavior(context);
1044 TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
1046 long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
1048 assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis());
1050 RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
1051 assertTrue("Expected Candidate", newBehavior instanceof Candidate);
1055 public void testFollowerSchedulesElectionIfAutomaticElectionsAreDisabled() {
1056 MockRaftActorContext context = createActorContext();
1057 context.setConfigParams(new DefaultConfigParamsImpl() {
1059 public FiniteDuration getElectionTimeOutInterval() {
1060 return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
1064 context.setRaftPolicy(createRaftPolicy(false, false));
1066 follower = createBehavior(context);
1068 TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
1069 RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
1070 assertSame("handleMessage result", follower, newBehavior);
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);
1081 follower = new Follower(context, "leader", (short)1);
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());
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;
1098 public long getTerm() {
1102 verify(follower).scheduleElection(any(FiniteDuration.class));
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));
1114 public void testCaptureSnapshotOnLastEntryInAppendEntries() throws Exception {
1115 String id = "testCaptureSnapshotOnLastEntryInAppendEntries";
1118 InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1120 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1121 config.setSnapshotBatchCount(2);
1122 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
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();
1133 InMemorySnapshotStore.addSnapshotSavedLatch(id);
1134 InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1135 InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
1137 List<ReplicatedLogEntry> entries = Arrays.asList(
1138 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"));
1140 AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 1, -1, (short)0);
1142 followerActorRef.tell(appendEntries, leaderActor);
1144 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1145 assertEquals("isSuccess", true, reply.isSuccess());
1147 final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
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());
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()));
1169 public void testCaptureSnapshotOnMiddleEntryInAppendEntries() throws Exception {
1170 String id = "testCaptureSnapshotOnMiddleEntryInAppendEntries";
1173 InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1175 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1176 config.setSnapshotBatchCount(2);
1177 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
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();
1188 InMemorySnapshotStore.addSnapshotSavedLatch(id);
1189 InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1190 InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
1192 List<ReplicatedLogEntry> entries = Arrays.asList(
1193 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
1194 newReplicatedLogEntry(1, 2, "three"));
1196 AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 2, -1, (short)0);
1198 followerActorRef.tell(appendEntries, leaderActor);
1200 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1201 assertEquals("isSuccess", true, reply.isSuccess());
1203 final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
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());
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()));
1223 assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
1224 assertEquals("Snapshot index", 2, followerRaftActor.get().getReplicatedLog().getSnapshotIndex());
1226 // Reinstate the actor from persistence
1228 actorFactory.killActor(followerActorRef, new JavaTestKit(getSystem()));
1230 followerActorRef = actorFactory.createTestActor(builder.props()
1231 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1232 followerRaftActor.set(followerActorRef.underlyingActor());
1233 followerRaftActor.get().waitForInitializeBehaviorComplete();
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());
1244 public void testCaptureSnapshotOnAppendEntriesWithUnapplied() throws Exception {
1245 String id = "testCaptureSnapshotOnAppendEntriesWithUnapplied";
1248 InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1250 DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1251 config.setSnapshotBatchCount(1);
1252 config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
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();
1263 InMemorySnapshotStore.addSnapshotSavedLatch(id);
1264 InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1265 InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
1267 List<ReplicatedLogEntry> entries = Arrays.asList(
1268 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
1269 newReplicatedLogEntry(1, 2, "three"));
1271 AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 0, -1, (short)0);
1273 followerActorRef.tell(appendEntries, leaderActor);
1275 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1276 assertEquals("isSuccess", true, reply.isSuccess());
1278 final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
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());
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()));
1301 @SuppressWarnings("checkstyle:IllegalCatch")
1302 private RaftActorSnapshotCohort newRaftActorSnapshotCohort(final AtomicReference<MockRaftActor> followerRaftActor) {
1303 RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() {
1305 public void createSnapshot(ActorRef actorRef, java.util.Optional<OutputStream> installSnapshotStream) {
1307 actorRef.tell(new CaptureSnapshotReply(new MockSnapshotState(followerRaftActor.get().getState()),
1308 installSnapshotStream), actorRef);
1309 } catch (Exception e) {
1310 Throwables.propagate(e);
1315 public void applySnapshot(State snapshotState) {
1319 public State deserializeSnapshot(ByteSource snapshotBytes) {
1320 throw new UnsupportedOperationException();
1323 return snapshotCohort;
1326 public byte[] getNextChunk(ByteString bs, int offset, int chunkSize) {
1327 int snapshotLength = bs.size();
1329 int size = chunkSize;
1330 if (chunkSize > snapshotLength) {
1331 size = snapshotLength;
1333 if (start + chunkSize > snapshotLength) {
1334 size = snapshotLength - start;
1338 byte[] nextChunk = new byte[size];
1339 bs.copyTo(nextChunk, start, 0, size);
1343 private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
1344 String expFollowerId, long expLogLastTerm, long expLogLastIndex) {
1345 expectAndVerifyAppendEntriesReply(expTerm, expSuccess, expFollowerId, expLogLastTerm, expLogLastIndex, false);
1348 private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
1349 String expFollowerId, long expLogLastTerm, long expLogLastIndex,
1350 boolean expForceInstallSnapshot) {
1352 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
1353 AppendEntriesReply.class);
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());
1365 private static ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
1366 return new SimpleReplicatedLogEntry(index, term,
1367 new MockRaftActorContext.MockPayload(data));
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");
1376 return toByteString(followerSnapshot);
1380 protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(MockRaftActorContext actorContext,
1381 ActorRef actorRef, RaftRPC rpc) throws Exception {
1382 super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
1384 String expVotedFor = rpc instanceof RequestVote ? ((RequestVote)rpc).getCandidateId() : null;
1385 assertEquals("New votedFor", expVotedFor, actorContext.getTermInformation().getVotedFor());
1389 protected void handleAppendEntriesAddSameEntryToLogReply(final TestActorRef<MessageCollectorActor> replyActor)
1391 AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(replyActor, AppendEntriesReply.class);
1392 assertEquals("isSuccess", true, reply.isSuccess());