opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/FollowerTest.java

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