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, null, 101, -1, (short)0);
 542
 543         follower = createBehavior(context);
 544
 545         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 546
 547         Assert.assertSame(follower, newBehavior);
 548
 549         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
 550                 AppendEntriesReply.class);
 551
 552         assertEquals("isSuccess", false, reply.isSuccess());
 553     }
 554
 555     /**
 556      * This test verifies that when a new AppendEntries message is received with
 557      * new entries and the logs of the sender and receiver match that the new
 558      * entries get added to the log and the log is incremented by the number of
 559      * entries received in appendEntries.
 560      */
 561     @Test
 562     public void testHandleAppendEntriesAddNewEntries() {
 563         logStart("testHandleAppendEntriesAddNewEntries");
 564
 565         MockRaftActorContext context = createActorContext();
 566
 567         // First set the receivers term to lower number
 568         context.getTermInformation().update(1, "test");
 569
 570         // Prepare the receivers log
 571         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 572         log.append(newReplicatedLogEntry(1, 0, "zero"));
 573         log.append(newReplicatedLogEntry(1, 1, "one"));
 574         log.append(newReplicatedLogEntry(1, 2, "two"));
 575
 576         context.setReplicatedLog(log);
 577
 578         // Prepare the entries to be sent with AppendEntries
 579         List<ReplicatedLogEntry> entries = new ArrayList<>();
 580         entries.add(newReplicatedLogEntry(1, 3, "three"));
 581         entries.add(newReplicatedLogEntry(1, 4, "four"));
 582
 583         // Send appendEntries with the same term as was set on the receiver
 584         // before the new behavior was created (1 in this case)
 585         // This will not work for a Candidate because as soon as a Candidate
 586         // is created it increments the term
 587         short leaderPayloadVersion = 10;
 588         String leaderId = "leader-1";
 589         AppendEntries appendEntries = new AppendEntries(1, leaderId, 2, 1, entries, 4, -1, leaderPayloadVersion);
 590
 591         follower = createBehavior(context);
 592
 593         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 594
 595         Assert.assertSame(follower, newBehavior);
 596
 597         assertEquals("Next index", 5, log.last().getIndex() + 1);
 598         assertEquals("Entry 3", entries.get(0), log.get(3));
 599         assertEquals("Entry 4", entries.get(1), log.get(4));
 600
 601         assertEquals("getLeaderPayloadVersion", leaderPayloadVersion, newBehavior.getLeaderPayloadVersion());
 602         assertEquals("getLeaderId", leaderId, newBehavior.getLeaderId());
 603
 604         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
 605     }
 606
 607     /**
 608      * This test verifies that when a new AppendEntries message is received with
 609      * new entries and the logs of the sender and receiver are out-of-sync that
 610      * the log is first corrected by removing the out of sync entries from the
 611      * log and then adding in the new entries sent with the AppendEntries message.
 612      */
 613     @Test
 614     public void testHandleAppendEntriesCorrectReceiverLogEntries() {
 615         logStart("testHandleAppendEntriesCorrectReceiverLogEntries");
 616
 617         MockRaftActorContext context = createActorContext();
 618
 619         // First set the receivers term to lower number
 620         context.getTermInformation().update(1, "test");
 621
 622         // Prepare the receivers log
 623         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 624         log.append(newReplicatedLogEntry(1, 0, "zero"));
 625         log.append(newReplicatedLogEntry(1, 1, "one"));
 626         log.append(newReplicatedLogEntry(1, 2, "two"));
 627
 628         context.setReplicatedLog(log);
 629
 630         // Prepare the entries to be sent with AppendEntries
 631         List<ReplicatedLogEntry> entries = new ArrayList<>();
 632         entries.add(newReplicatedLogEntry(2, 2, "two-1"));
 633         entries.add(newReplicatedLogEntry(2, 3, "three"));
 634
 635         // Send appendEntries with the same term as was set on the receiver
 636         // before the new behavior was created (1 in this case)
 637         // This will not work for a Candidate because as soon as a Candidate
 638         // is created it increments the term
 639         AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1, (short)0);
 640
 641         follower = createBehavior(context);
 642
 643         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 644
 645         Assert.assertSame(follower, newBehavior);
 646
 647         // The entry at index 2 will be found out-of-sync with the leader
 648         // and will be removed
 649         // Then the two new entries will be added to the log
 650         // Thus making the log to have 4 entries
 651         assertEquals("Next index", 4, log.last().getIndex() + 1);
 652         //assertEquals("Entry 2", entries.get(0), log.get(2));
 653
 654         assertEquals("Entry 1 data", "one", log.get(1).getData().toString());
 655
 656         // Check that the entry at index 2 has the new data
 657         assertEquals("Entry 2", entries.get(0), log.get(2));
 658
 659         assertEquals("Entry 3", entries.get(1), log.get(3));
 660
 661         expectAndVerifyAppendEntriesReply(2, true, context.getId(), 2, 3);
 662     }
 663
 664     @Test
 665     public void testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot() {
 666         logStart("testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot");
 667
 668         MockRaftActorContext context = createActorContext();
 669
 670         // First set the receivers term to lower number
 671         context.getTermInformation().update(1, "test");
 672
 673         // Prepare the receivers log
 674         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 675         log.append(newReplicatedLogEntry(1, 0, "zero"));
 676         log.append(newReplicatedLogEntry(1, 1, "one"));
 677         log.append(newReplicatedLogEntry(1, 2, "two"));
 678
 679         context.setReplicatedLog(log);
 680
 681         // Prepare the entries to be sent with AppendEntries
 682         List<ReplicatedLogEntry> entries = new ArrayList<>();
 683         entries.add(newReplicatedLogEntry(2, 2, "two-1"));
 684         entries.add(newReplicatedLogEntry(2, 3, "three"));
 685
 686         // Send appendEntries with the same term as was set on the receiver
 687         // before the new behavior was created (1 in this case)
 688         // This will not work for a Candidate because as soon as a Candidate
 689         // is created it increments the term
 690         AppendEntries appendEntries = new AppendEntries(2, "leader", 1, 1, entries, 3, -1, (short)0);
 691
 692         context.setRaftPolicy(createRaftPolicy(false, true));
 693         follower = createBehavior(context);
 694
 695         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 696
 697         Assert.assertSame(follower, newBehavior);
 698
 699         expectAndVerifyAppendEntriesReply(2, false, context.getId(), 1, 2, true);
 700     }
 701
 702     @Test
 703     public void testHandleAppendEntriesPreviousLogEntryMissing() {
 704         logStart("testHandleAppendEntriesPreviousLogEntryMissing");
 705
 706         final MockRaftActorContext context = createActorContext();
 707
 708         // Prepare the receivers log
 709         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 710         log.append(newReplicatedLogEntry(1, 0, "zero"));
 711         log.append(newReplicatedLogEntry(1, 1, "one"));
 712         log.append(newReplicatedLogEntry(1, 2, "two"));
 713
 714         context.setReplicatedLog(log);
 715
 716         // Prepare the entries to be sent with AppendEntries
 717         List<ReplicatedLogEntry> entries = new ArrayList<>();
 718         entries.add(newReplicatedLogEntry(1, 4, "four"));
 719
 720         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1, (short)0);
 721
 722         follower = createBehavior(context);
 723
 724         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 725
 726         Assert.assertSame(follower, newBehavior);
 727
 728         expectAndVerifyAppendEntriesReply(1, false, context.getId(), 1, 2);
 729     }
 730
 731     @Test
 732     public void testHandleAppendEntriesWithExistingLogEntry() {
 733         logStart("testHandleAppendEntriesWithExistingLogEntry");
 734
 735         MockRaftActorContext context = createActorContext();
 736
 737         context.getTermInformation().update(1, "test");
 738
 739         // Prepare the receivers log
 740         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 741         log.append(newReplicatedLogEntry(1, 0, "zero"));
 742         log.append(newReplicatedLogEntry(1, 1, "one"));
 743
 744         context.setReplicatedLog(log);
 745
 746         // Send the last entry again.
 747         List<ReplicatedLogEntry> entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"));
 748
 749         follower = createBehavior(context);
 750
 751         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1, (short)0));
 752
 753         assertEquals("Next index", 2, log.last().getIndex() + 1);
 754         assertEquals("Entry 1", entries.get(0), log.get(1));
 755
 756         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 1);
 757
 758         // Send the last entry again and also a new one.
 759
 760         entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two"));
 761
 762         leaderActor.underlyingActor().clear();
 763         follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1, (short)0));
 764
 765         assertEquals("Next index", 3, log.last().getIndex() + 1);
 766         assertEquals("Entry 1", entries.get(0), log.get(1));
 767         assertEquals("Entry 2", entries.get(1), log.get(2));
 768
 769         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 2);
 770     }
 771
 772     @Test
 773     public void testHandleAppendEntriesAfterInstallingSnapshot() {
 774         logStart("testHandleAppendAfterInstallingSnapshot");
 775
 776         MockRaftActorContext context = createActorContext();
 777
 778         // Prepare the receivers log
 779         MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
 780
 781         // Set up a log as if it has been snapshotted
 782         log.setSnapshotIndex(3);
 783         log.setSnapshotTerm(1);
 784
 785         context.setReplicatedLog(log);
 786
 787         // Prepare the entries to be sent with AppendEntries
 788         List<ReplicatedLogEntry> entries = new ArrayList<>();
 789         entries.add(newReplicatedLogEntry(1, 4, "four"));
 790
 791         AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3, (short)0);
 792
 793         follower = createBehavior(context);
 794
 795         RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
 796
 797         Assert.assertSame(follower, newBehavior);
 798
 799         expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
 800     }
 801
 802
 803     /**
 804      * This test verifies that when InstallSnapshot is received by
 805      * the follower its applied correctly.
 806      */
 807     @Test
 808     public void testHandleInstallSnapshot() throws Exception {
 809         logStart("testHandleInstallSnapshot");
 810
 811         MockRaftActorContext context = createActorContext();
 812         context.getTermInformation().update(1, "leader");
 813
 814         follower = createBehavior(context);
 815
 816         ByteString bsSnapshot = createSnapshot();
 817         int offset = 0;
 818         int snapshotLength = bsSnapshot.size();
 819         int chunkSize = 50;
 820         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
 821         int lastIncludedIndex = 1;
 822         int chunkIndex = 1;
 823         InstallSnapshot lastInstallSnapshot = null;
 824
 825         for (int i = 0; i < totalChunks; i++) {
 826             byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
 827             lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
 828                     chunkData, chunkIndex, totalChunks);
 829             follower.handleMessage(leaderActor, lastInstallSnapshot);
 830             offset = offset + 50;
 831             lastIncludedIndex++;
 832             chunkIndex++;
 833         }
 834
 835         ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
 836                 ApplySnapshot.class);
 837         Snapshot snapshot = applySnapshot.getSnapshot();
 838         assertNotNull(lastInstallSnapshot);
 839         assertEquals("getLastIndex", lastInstallSnapshot.getLastIncludedIndex(), snapshot.getLastIndex());
 840         assertEquals("getLastIncludedTerm", lastInstallSnapshot.getLastIncludedTerm(),
 841                 snapshot.getLastAppliedTerm());
 842         assertEquals("getLastAppliedIndex", lastInstallSnapshot.getLastIncludedIndex(),
 843                 snapshot.getLastAppliedIndex());
 844         assertEquals("getLastTerm", lastInstallSnapshot.getLastIncludedTerm(), snapshot.getLastTerm());
 845         assertEquals("getState type", ByteState.class, snapshot.getState().getClass());
 846         Assert.assertArrayEquals("getState", bsSnapshot.toByteArray(), ((ByteState)snapshot.getState()).getBytes());
 847         assertEquals("getElectionTerm", 1, snapshot.getElectionTerm());
 848         assertEquals("getElectionVotedFor", "leader", snapshot.getElectionVotedFor());
 849         applySnapshot.getCallback().onSuccess();
 850
 851         List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
 852                 leaderActor, InstallSnapshotReply.class);
 853         assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
 854
 855         chunkIndex = 1;
 856         for (InstallSnapshotReply reply: replies) {
 857             assertEquals("getChunkIndex", chunkIndex++, reply.getChunkIndex());
 858             assertEquals("getTerm", 1, reply.getTerm());
 859             assertEquals("isSuccess", true, reply.isSuccess());
 860             assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
 861         }
 862
 863         assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
 864     }
 865
 866
 867     /**
 868      * Verify that when an AppendEntries is sent to a follower during a snapshot install
 869      * the Follower short-circuits the processing of the AppendEntries message.
 870      */
 871     @Test
 872     public void testReceivingAppendEntriesDuringInstallSnapshot() throws Exception {
 873         logStart("testReceivingAppendEntriesDuringInstallSnapshot");
 874
 875         MockRaftActorContext context = createActorContext();
 876
 877         follower = createBehavior(context);
 878
 879         ByteString bsSnapshot  = createSnapshot();
 880         int snapshotLength = bsSnapshot.size();
 881         int chunkSize = 50;
 882         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
 883         int lastIncludedIndex = 1;
 884
 885         // Check that snapshot installation is not in progress
 886         assertNull(follower.getSnapshotTracker());
 887
 888         // Make sure that we have more than 1 chunk to send
 889         assertTrue(totalChunks > 1);
 890
 891         // Send an install snapshot with the first chunk to start the process of installing a snapshot
 892         byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
 893         follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
 894                 chunkData, 1, totalChunks));
 895
 896         // Check if snapshot installation is in progress now
 897         assertNotNull(follower.getSnapshotTracker());
 898
 899         // Send an append entry
 900         AppendEntries appendEntries = new AppendEntries(1, "leader", 1, 1,
 901                 Arrays.asList(newReplicatedLogEntry(2, 1, "3")), 2, -1, (short)1);
 902
 903         follower.handleMessage(leaderActor, appendEntries);
 904
 905         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
 906         assertEquals("isSuccess", true, reply.isSuccess());
 907         assertEquals("getLogLastIndex", context.getReplicatedLog().lastIndex(), reply.getLogLastIndex());
 908         assertEquals("getLogLastTerm", context.getReplicatedLog().lastTerm(), reply.getLogLastTerm());
 909         assertEquals("getTerm", context.getTermInformation().getCurrentTerm(), reply.getTerm());
 910
 911         assertNotNull(follower.getSnapshotTracker());
 912     }
 913
 914     @Test
 915     public void testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader() throws Exception {
 916         logStart("testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader");
 917
 918         MockRaftActorContext context = createActorContext();
 919
 920         follower = createBehavior(context);
 921
 922         ByteString bsSnapshot  = createSnapshot();
 923         int snapshotLength = bsSnapshot.size();
 924         int chunkSize = 50;
 925         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
 926         int lastIncludedIndex = 1;
 927
 928         // Check that snapshot installation is not in progress
 929         assertNull(follower.getSnapshotTracker());
 930
 931         // Make sure that we have more than 1 chunk to send
 932         assertTrue(totalChunks > 1);
 933
 934         // Send an install snapshot with the first chunk to start the process of installing a snapshot
 935         byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
 936         follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
 937                 chunkData, 1, totalChunks));
 938
 939         // Check if snapshot installation is in progress now
 940         assertNotNull(follower.getSnapshotTracker());
 941
 942         // Send appendEntries with a new term and leader.
 943         AppendEntries appendEntries = new AppendEntries(2, "new-leader", 1, 1,
 944                 Arrays.asList(newReplicatedLogEntry(2, 2, "3")), 2, -1, (short)1);
 945
 946         follower.handleMessage(leaderActor, appendEntries);
 947
 948         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
 949         assertEquals("isSuccess", true, reply.isSuccess());
 950         assertEquals("getLogLastIndex", 2, reply.getLogLastIndex());
 951         assertEquals("getLogLastTerm", 2, reply.getLogLastTerm());
 952         assertEquals("getTerm", 2, reply.getTerm());
 953
 954         assertNull(follower.getSnapshotTracker());
 955     }
 956
 957     @Test
 958     public void testInitialSyncUpWithHandleInstallSnapshotFollowedByAppendEntries() throws Exception {
 959         logStart("testInitialSyncUpWithHandleInstallSnapshot");
 960
 961         MockRaftActorContext context = createActorContext();
 962         context.setCommitIndex(-1);
 963
 964         follower = createBehavior(context);
 965
 966         ByteString bsSnapshot  = createSnapshot();
 967         int offset = 0;
 968         int snapshotLength = bsSnapshot.size();
 969         int chunkSize = 50;
 970         int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
 971         int lastIncludedIndex = 1;
 972         int chunkIndex = 1;
 973         InstallSnapshot lastInstallSnapshot = null;
 974
 975         for (int i = 0; i < totalChunks; i++) {
 976             byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
 977             lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
 978                     chunkData, chunkIndex, totalChunks);
 979             follower.handleMessage(leaderActor, lastInstallSnapshot);
 980             offset = offset + 50;
 981             lastIncludedIndex++;
 982             chunkIndex++;
 983         }
 984
 985         FollowerInitialSyncUpStatus syncStatus =
 986                 MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
 987
 988         assertFalse(syncStatus.isInitialSyncDone());
 989
 990         // Clear all the messages
 991         followerActor.underlyingActor().clear();
 992
 993         context.setLastApplied(101);
 994         context.setCommitIndex(101);
 995         setLastLogEntry(context, 1, 101,
 996                 new MockRaftActorContext.MockPayload(""));
 997
 998         List<ReplicatedLogEntry> entries = Arrays.asList(
 999                 newReplicatedLogEntry(2, 101, "foo"));
1000
1001         // The new commitIndex is 101
1002         AppendEntries appendEntries = new AppendEntries(2, "leader", 101, 1, entries, 102, 101, (short)0);
1003         follower.handleMessage(leaderActor, appendEntries);
1004
1005         syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
1006
1007         assertTrue(syncStatus.isInitialSyncDone());
1008     }
1009
1010     @Test
1011     public void testHandleOutOfSequenceInstallSnapshot() throws Exception {
1012         logStart("testHandleOutOfSequenceInstallSnapshot");
1013
1014         MockRaftActorContext context = createActorContext();
1015
1016         follower = createBehavior(context);
1017
1018         ByteString bsSnapshot = createSnapshot();
1019
1020         InstallSnapshot installSnapshot = new InstallSnapshot(1, "leader", 3, 1,
1021                 getNextChunk(bsSnapshot, 10, 50), 3, 3);
1022         follower.handleMessage(leaderActor, installSnapshot);
1023
1024         InstallSnapshotReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
1025                 InstallSnapshotReply.class);
1026
1027         assertEquals("isSuccess", false, reply.isSuccess());
1028         assertEquals("getChunkIndex", -1, reply.getChunkIndex());
1029         assertEquals("getTerm", 1, reply.getTerm());
1030         assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
1031
1032         assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
1033     }
1034
1035     @Test
1036     public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers() {
1037         MockRaftActorContext context = createActorContext();
1038
1039         Stopwatch stopwatch = Stopwatch.createStarted();
1040
1041         follower = createBehavior(context);
1042
1043         TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
1044
1045         long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
1046
1047         assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis());
1048
1049         RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
1050         assertTrue("Expected Candidate", newBehavior instanceof Candidate);
1051     }
1052
1053     @Test
1054     public void testFollowerSchedulesElectionIfAutomaticElectionsAreDisabled() {
1055         MockRaftActorContext context = createActorContext();
1056         context.setConfigParams(new DefaultConfigParamsImpl() {
1057             @Override
1058             public FiniteDuration getElectionTimeOutInterval() {
1059                 return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
1060             }
1061         });
1062
1063         context.setRaftPolicy(createRaftPolicy(false, false));
1064
1065         follower = createBehavior(context);
1066
1067         TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
1068         RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
1069         assertSame("handleMessage result", follower, newBehavior);
1070     }
1071
1072     @Test
1073     public void testFollowerSchedulesElectionIfNonVoting() {
1074         MockRaftActorContext context = createActorContext();
1075         context.updatePeerIds(new ServerConfigurationPayload(Arrays.asList(new ServerInfo(context.getId(), false))));
1076         ((DefaultConfigParamsImpl)context.getConfigParams()).setHeartBeatInterval(
1077                 FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
1078         ((DefaultConfigParamsImpl)context.getConfigParams()).setElectionTimeoutFactor(1);
1079
1080         follower = new Follower(context, "leader", (short)1);
1081
1082         ElectionTimeout electionTimeout = MessageCollectorActor.expectFirstMatching(followerActor,
1083                 ElectionTimeout.class);
1084         RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), electionTimeout);
1085         assertSame("handleMessage result", follower, newBehavior);
1086         assertNull("Expected null leaderId", follower.getLeaderId());
1087     }
1088
1089     @Test
1090     public void testElectionScheduledWhenAnyRaftRPCReceived() {
1091         MockRaftActorContext context = createActorContext();
1092         follower = createBehavior(context);
1093         follower.handleMessage(leaderActor, new RaftRPC() {
1094             private static final long serialVersionUID = 1L;
1095
1096             @Override
1097             public long getTerm() {
1098                 return 100;
1099             }
1100         });
1101         verify(follower).scheduleElection(any(FiniteDuration.class));
1102     }
1103
1104     @Test
1105     public void testElectionNotScheduledWhenNonRaftRPCMessageReceived() {
1106         MockRaftActorContext context = createActorContext();
1107         follower = createBehavior(context);
1108         follower.handleMessage(leaderActor, "non-raft-rpc");
1109         verify(follower, never()).scheduleElection(any(FiniteDuration.class));
1110     }
1111
1112     @Test
1113     public void testCaptureSnapshotOnLastEntryInAppendEntries() throws Exception {
1114         String id = "testCaptureSnapshotOnLastEntryInAppendEntries";
1115         logStart(id);
1116
1117         InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1118
1119         DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1120         config.setSnapshotBatchCount(2);
1121         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1122
1123         final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
1124         RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
1125         Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
1126                 .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
1127         TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
1128                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1129         followerRaftActor.set(followerActorRef.underlyingActor());
1130         followerRaftActor.get().waitForInitializeBehaviorComplete();
1131
1132         InMemorySnapshotStore.addSnapshotSavedLatch(id);
1133         InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1134
1135         List<ReplicatedLogEntry> entries = Arrays.asList(
1136                 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"));
1137
1138         AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 1, -1, (short)0);
1139
1140         followerActorRef.tell(appendEntries, leaderActor);
1141
1142         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1143         assertEquals("isSuccess", true, reply.isSuccess());
1144
1145         final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
1146
1147         InMemoryJournal.waitForDeleteMessagesComplete(id);
1148         // We expect the ApplyJournalEntries for index 1 to remain in the persisted log b/c it's still queued for
1149         // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
1150         // This is OK - on recovery it will be a no-op since index 1 has already been applied.
1151         List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
1152         assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
1153         assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
1154         assertEquals("ApplyJournalEntries index", 1, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
1155
1156         assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
1157         assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
1158         assertEquals("Snapshot getLastAppliedIndex", 1, snapshot.getLastAppliedIndex());
1159         assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
1160         assertEquals("Snapshot getLastIndex", 1, snapshot.getLastIndex());
1161         assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData()),
1162                 MockRaftActor.fromState(snapshot.getState()));
1163     }
1164
1165     @Test
1166     public void testCaptureSnapshotOnMiddleEntryInAppendEntries() throws Exception {
1167         String id = "testCaptureSnapshotOnMiddleEntryInAppendEntries";
1168         logStart(id);
1169
1170         InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1171
1172         DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1173         config.setSnapshotBatchCount(2);
1174         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1175
1176         final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
1177         RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
1178         Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
1179                 .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
1180         TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
1181                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1182         followerRaftActor.set(followerActorRef.underlyingActor());
1183         followerRaftActor.get().waitForInitializeBehaviorComplete();
1184
1185         InMemorySnapshotStore.addSnapshotSavedLatch(id);
1186         InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1187
1188         List<ReplicatedLogEntry> entries = Arrays.asList(
1189                 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
1190                 newReplicatedLogEntry(1, 2, "three"));
1191
1192         AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 2, -1, (short)0);
1193
1194         followerActorRef.tell(appendEntries, leaderActor);
1195
1196         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1197         assertEquals("isSuccess", true, reply.isSuccess());
1198
1199         final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
1200
1201         InMemoryJournal.waitForDeleteMessagesComplete(id);
1202         // We expect the ApplyJournalEntries for index 2 to remain in the persisted log b/c it's still queued for
1203         // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
1204         // This is OK - on recovery it will be a no-op since index 2 has already been applied.
1205         List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
1206         assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
1207         assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
1208         assertEquals("ApplyJournalEntries index", 2, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
1209
1210         assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
1211         assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
1212         assertEquals("Snapshot getLastAppliedIndex", 2, snapshot.getLastAppliedIndex());
1213         assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
1214         assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
1215         assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
1216                 entries.get(2).getData()), MockRaftActor.fromState(snapshot.getState()));
1217
1218         assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
1219         assertEquals("Snapshot index", 2, followerRaftActor.get().getReplicatedLog().getSnapshotIndex());
1220
1221         // Reinstate the actor from persistence
1222
1223         actorFactory.killActor(followerActorRef, new JavaTestKit(getSystem()));
1224
1225         followerActorRef = actorFactory.createTestActor(builder.props()
1226                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1227         followerRaftActor.set(followerActorRef.underlyingActor());
1228         followerRaftActor.get().waitForInitializeBehaviorComplete();
1229
1230         assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
1231         assertEquals("Last index", 2, followerRaftActor.get().getReplicatedLog().lastIndex());
1232         assertEquals("Last applied index", 2, followerRaftActor.get().getRaftActorContext().getLastApplied());
1233         assertEquals("Commit index", 2, followerRaftActor.get().getRaftActorContext().getCommitIndex());
1234         assertEquals("State", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
1235                 entries.get(2).getData()), followerRaftActor.get().getState());
1236     }
1237
1238     @Test
1239     public void testCaptureSnapshotOnAppendEntriesWithUnapplied() throws Exception {
1240         String id = "testCaptureSnapshotOnAppendEntriesWithUnapplied";
1241         logStart(id);
1242
1243         InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
1244
1245         DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
1246         config.setSnapshotBatchCount(1);
1247         config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
1248
1249         final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
1250         RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
1251         Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
1252                 .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
1253         TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
1254                 .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
1255         followerRaftActor.set(followerActorRef.underlyingActor());
1256         followerRaftActor.get().waitForInitializeBehaviorComplete();
1257
1258         InMemorySnapshotStore.addSnapshotSavedLatch(id);
1259         InMemoryJournal.addDeleteMessagesCompleteLatch(id);
1260
1261         List<ReplicatedLogEntry> entries = Arrays.asList(
1262                 newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
1263                 newReplicatedLogEntry(1, 2, "three"));
1264
1265         AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 0, -1, (short)0);
1266
1267         followerActorRef.tell(appendEntries, leaderActor);
1268
1269         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1270         assertEquals("isSuccess", true, reply.isSuccess());
1271
1272         final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
1273
1274         InMemoryJournal.waitForDeleteMessagesComplete(id);
1275         // We expect the ApplyJournalEntries for index 0 to remain in the persisted log b/c it's still queued for
1276         // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
1277         // This is OK - on recovery it will be a no-op since index 0 has already been applied.
1278         List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
1279         assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
1280         assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
1281         assertEquals("ApplyJournalEntries index", 0, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
1282
1283         assertEquals("Snapshot unapplied size", 2, snapshot.getUnAppliedEntries().size());
1284         assertEquals("Snapshot unapplied entry index", 1, snapshot.getUnAppliedEntries().get(0).getIndex());
1285         assertEquals("Snapshot unapplied entry index", 2, snapshot.getUnAppliedEntries().get(1).getIndex());
1286         assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
1287         assertEquals("Snapshot getLastAppliedIndex", 0, snapshot.getLastAppliedIndex());
1288         assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
1289         assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
1290         assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData()),
1291                 MockRaftActor.fromState(snapshot.getState()));
1292     }
1293
1294     @SuppressWarnings("checkstyle:IllegalCatch")
1295     private RaftActorSnapshotCohort newRaftActorSnapshotCohort(final AtomicReference<MockRaftActor> followerRaftActor) {
1296         RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() {
1297             @Override
1298             public void createSnapshot(ActorRef actorRef, java.util.Optional<OutputStream> installSnapshotStream) {
1299                 try {
1300                     actorRef.tell(new CaptureSnapshotReply(new MockSnapshotState(followerRaftActor.get().getState()),
1301                             installSnapshotStream), actorRef);
1302                 } catch (Exception e) {
1303                     Throwables.propagate(e);
1304                 }
1305             }
1306
1307             @Override
1308             public void applySnapshot(State snapshotState) {
1309             }
1310
1311             @Override
1312             public State deserializeSnapshot(ByteSource snapshotBytes) {
1313                 throw new UnsupportedOperationException();
1314             }
1315         };
1316         return snapshotCohort;
1317     }
1318
1319     public byte[] getNextChunk(ByteString bs, int offset, int chunkSize) {
1320         int snapshotLength = bs.size();
1321         int start = offset;
1322         int size = chunkSize;
1323         if (chunkSize > snapshotLength) {
1324             size = snapshotLength;
1325         } else {
1326             if (start + chunkSize > snapshotLength) {
1327                 size = snapshotLength - start;
1328             }
1329         }
1330
1331         byte[] nextChunk = new byte[size];
1332         bs.copyTo(nextChunk, start, 0, size);
1333         return nextChunk;
1334     }
1335
1336     private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
1337             String expFollowerId, long expLogLastTerm, long expLogLastIndex) {
1338         expectAndVerifyAppendEntriesReply(expTerm, expSuccess, expFollowerId, expLogLastTerm, expLogLastIndex, false);
1339     }
1340
1341     private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
1342                                                    String expFollowerId, long expLogLastTerm, long expLogLastIndex,
1343                                                    boolean expForceInstallSnapshot) {
1344
1345         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
1346                 AppendEntriesReply.class);
1347
1348         assertEquals("isSuccess", expSuccess, reply.isSuccess());
1349         assertEquals("getTerm", expTerm, reply.getTerm());
1350         assertEquals("getFollowerId", expFollowerId, reply.getFollowerId());
1351         assertEquals("getLogLastTerm", expLogLastTerm, reply.getLogLastTerm());
1352         assertEquals("getLogLastIndex", expLogLastIndex, reply.getLogLastIndex());
1353         assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion());
1354         assertEquals("isForceInstallSnapshot", expForceInstallSnapshot, reply.isForceInstallSnapshot());
1355     }
1356
1357
1358     private static ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
1359         return new SimpleReplicatedLogEntry(index, term,
1360                 new MockRaftActorContext.MockPayload(data));
1361     }
1362
1363     private ByteString createSnapshot() {
1364         HashMap<String, String> followerSnapshot = new HashMap<>();
1365         followerSnapshot.put("1", "A");
1366         followerSnapshot.put("2", "B");
1367         followerSnapshot.put("3", "C");
1368
1369         return toByteString(followerSnapshot);
1370     }
1371
1372     @Override
1373     protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(MockRaftActorContext actorContext,
1374             ActorRef actorRef, RaftRPC rpc) throws Exception {
1375         super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
1376
1377         String expVotedFor = rpc instanceof RequestVote ? ((RequestVote)rpc).getCandidateId() : null;
1378         assertEquals("New votedFor", expVotedFor, actorContext.getTermInformation().getVotedFor());
1379     }
1380
1381     @Override
1382     protected void handleAppendEntriesAddSameEntryToLogReply(final TestActorRef<MessageCollectorActor> replyActor)
1383             throws Exception {
1384         AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(replyActor, AppendEntriesReply.class);
1385         assertEquals("isSuccess", true, reply.isSuccess());
1386     }
1387 }