opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/LeaderTest.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.Mockito.mock;
  18 import static org.mockito.Mockito.never;
  19 import static org.mockito.Mockito.verify;
  20
  21 import akka.actor.ActorRef;
  22 import akka.actor.PoisonPill;
  23 import akka.actor.Props;
  24 import akka.actor.Terminated;
  25 import akka.testkit.JavaTestKit;
  26 import akka.testkit.TestActorRef;
  27 import com.google.common.collect.ImmutableMap;
  28 import com.google.common.util.concurrent.Uninterruptibles;
  29 import com.google.protobuf.ByteString;
  30 import java.util.Arrays;
  31 import java.util.Collections;
  32 import java.util.HashMap;
  33 import java.util.List;
  34 import java.util.Map;
  35 import java.util.concurrent.TimeUnit;
  36 import org.junit.After;
  37 import org.junit.Test;
  38 import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
  39 import org.opendaylight.controller.cluster.raft.FollowerLogInformation;
  40 import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
  41 import org.opendaylight.controller.cluster.raft.RaftActorContext;
  42 import org.opendaylight.controller.cluster.raft.RaftActorLeadershipTransferCohort;
  43 import org.opendaylight.controller.cluster.raft.RaftState;
  44 import org.opendaylight.controller.cluster.raft.RaftVersions;
  45 import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
  46 import org.opendaylight.controller.cluster.raft.Snapshot;
  47 import org.opendaylight.controller.cluster.raft.VotingState;
  48 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
  49 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
  50 import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
  51 import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
  52 import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
  53 import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
  54 import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
  55 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  56 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  57 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
  58 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
  59 import org.opendaylight.controller.cluster.raft.messages.RaftRPC;
  60 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  61 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
  62 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
  63 import org.opendaylight.controller.cluster.raft.policy.DefaultRaftPolicy;
  64 import org.opendaylight.controller.cluster.raft.policy.RaftPolicy;
  65 import org.opendaylight.controller.cluster.raft.utils.ForwardMessageToBehaviorActor;
  66 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  67 import org.opendaylight.yangtools.concepts.Identifier;
  68 import scala.concurrent.duration.FiniteDuration;
  69
  70 public class LeaderTest extends AbstractLeaderTest<Leader> {
  71
  72     static final String FOLLOWER_ID = "follower";
  73     public static final String LEADER_ID = "leader";
  74
  75     private final TestActorRef<ForwardMessageToBehaviorActor> leaderActor = actorFactory.createTestActor(
  76             Props.create(ForwardMessageToBehaviorActor.class), actorFactory.generateActorId("leader"));
  77
  78     private final TestActorRef<ForwardMessageToBehaviorActor> followerActor = actorFactory.createTestActor(
  79             Props.create(ForwardMessageToBehaviorActor.class), actorFactory.generateActorId("follower"));
  80
  81     private Leader leader;
  82     private final short payloadVersion = 5;
  83
  84     @Override
  85     @After
  86     public void tearDown() throws Exception {
  87         if (leader != null) {
  88             leader.close();
  89         }
  90
  91         super.tearDown();
  92     }
  93
  94     @Test
  95     public void testHandleMessageForUnknownMessage() throws Exception {
  96         logStart("testHandleMessageForUnknownMessage");
  97
  98         leader = new Leader(createActorContext());
  99
 100         // handle message should null when it receives an unknown message
 101         assertNull(leader.handleMessage(followerActor, "foo"));
 102     }
 103
 104     @Test
 105     public void testThatLeaderSendsAHeartbeatMessageToAllFollowers() throws Exception {
 106         logStart("testThatLeaderSendsAHeartbeatMessageToAllFollowers");
 107
 108         MockRaftActorContext actorContext = createActorContextWithFollower();
 109         actorContext.setCommitIndex(-1);
 110         actorContext.setPayloadVersion(payloadVersion);
 111
 112         long term = 1;
 113         actorContext.getTermInformation().update(term, "");
 114
 115         leader = new Leader(actorContext);
 116         actorContext.setCurrentBehavior(leader);
 117
 118         // Leader should send an immediate heartbeat with no entries as follower is inactive.
 119         final long lastIndex = actorContext.getReplicatedLog().lastIndex();
 120         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 121         assertEquals("getTerm", term, appendEntries.getTerm());
 122         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
 123         assertEquals("getPrevLogTerm", -1, appendEntries.getPrevLogTerm());
 124         assertEquals("Entries size", 0, appendEntries.getEntries().size());
 125         assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion());
 126
 127         // The follower would normally reply - simulate that explicitly here.
 128         leader.handleMessage(followerActor, new AppendEntriesReply(
 129                 FOLLOWER_ID, term, true, lastIndex - 1, term, (short)0));
 130         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 131
 132         followerActor.underlyingActor().clear();
 133
 134         // Sleep for the heartbeat interval so AppendEntries is sent.
 135         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams()
 136                 .getHeartBeatInterval().toMillis(), TimeUnit.MILLISECONDS);
 137
 138         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
 139
 140         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 141         assertEquals("getPrevLogIndex", lastIndex - 1, appendEntries.getPrevLogIndex());
 142         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
 143         assertEquals("Entries size", 1, appendEntries.getEntries().size());
 144         assertEquals("Entry getIndex", lastIndex, appendEntries.getEntries().get(0).getIndex());
 145         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
 146         assertEquals("getPayloadVersion", payloadVersion, appendEntries.getPayloadVersion());
 147     }
 148
 149
 150     private RaftActorBehavior sendReplicate(MockRaftActorContext actorContext, long index) {
 151         return sendReplicate(actorContext, 1, index);
 152     }
 153
 154     private RaftActorBehavior sendReplicate(MockRaftActorContext actorContext, long term, long index) {
 155         MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("foo");
 156         SimpleReplicatedLogEntry newEntry = new SimpleReplicatedLogEntry(index, term, payload);
 157         actorContext.getReplicatedLog().append(newEntry);
 158         return leader.handleMessage(leaderActor, new Replicate(null, null, newEntry));
 159     }
 160
 161     @Test
 162     public void testHandleReplicateMessageSendAppendEntriesToFollower() throws Exception {
 163         logStart("testHandleReplicateMessageSendAppendEntriesToFollower");
 164
 165         MockRaftActorContext actorContext = createActorContextWithFollower();
 166
 167         long term = 1;
 168         actorContext.getTermInformation().update(term, "");
 169
 170         leader = new Leader(actorContext);
 171
 172         // Leader will send an immediate heartbeat - ignore it.
 173         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 174
 175         // The follower would normally reply - simulate that explicitly here.
 176         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 177         leader.handleMessage(followerActor, new AppendEntriesReply(
 178                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
 179         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 180
 181         followerActor.underlyingActor().clear();
 182
 183         RaftActorBehavior raftBehavior = sendReplicate(actorContext, lastIndex + 1);
 184
 185         // State should not change
 186         assertTrue(raftBehavior instanceof Leader);
 187
 188         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 189         assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex());
 190         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
 191         assertEquals("Entries size", 1, appendEntries.getEntries().size());
 192         assertEquals("Entry getIndex", lastIndex + 1, appendEntries.getEntries().get(0).getIndex());
 193         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
 194         assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString());
 195         assertEquals("Commit Index", lastIndex, actorContext.getCommitIndex());
 196     }
 197
 198     @Test
 199     public void testHandleReplicateMessageWithHigherTermThanPreviousEntry() throws Exception {
 200         logStart("testHandleReplicateMessageWithHigherTermThanPreviousEntry");
 201
 202         MockRaftActorContext actorContext = createActorContextWithFollower();
 203         actorContext.setCommitIndex(-1);
 204         actorContext.setLastApplied(-1);
 205
 206         // The raft context is initialized with a couple log entries. However the commitIndex
 207         // is -1, simulating that the leader previously didn't get consensus and thus the log entries weren't
 208         // committed and applied. Now it regains leadership with a higher term (2).
 209         long prevTerm = actorContext.getTermInformation().getCurrentTerm();
 210         long newTerm = prevTerm + 1;
 211         actorContext.getTermInformation().update(newTerm, "");
 212
 213         leader = new Leader(actorContext);
 214         actorContext.setCurrentBehavior(leader);
 215
 216         // Leader will send an immediate heartbeat - ignore it.
 217         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 218
 219         // The follower replies with the leader's current last index and term, simulating that it is
 220         // up to date with the leader.
 221         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 222         leader.handleMessage(followerActor, new AppendEntriesReply(
 223                 FOLLOWER_ID, newTerm, true, lastIndex, prevTerm, (short)0));
 224
 225         // The commit index should not get updated even though consensus was reached. This is b/c the
 226         // last entry's term does match the current term. As per §5.4.1, "Raft never commits log entries
 227         // from previous terms by counting replicas".
 228         assertEquals("Commit Index", -1, actorContext.getCommitIndex());
 229
 230         followerActor.underlyingActor().clear();
 231
 232         // Now replicate a new entry with the new term 2.
 233         long newIndex = lastIndex + 1;
 234         sendReplicate(actorContext, newTerm, newIndex);
 235
 236         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 237         assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex());
 238         assertEquals("getPrevLogTerm", prevTerm, appendEntries.getPrevLogTerm());
 239         assertEquals("Entries size", 1, appendEntries.getEntries().size());
 240         assertEquals("Entry getIndex", newIndex, appendEntries.getEntries().get(0).getIndex());
 241         assertEquals("Entry getTerm", newTerm, appendEntries.getEntries().get(0).getTerm());
 242         assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString());
 243
 244         // The follower replies with success. The leader should now update the commit index to the new index
 245         // as per §5.4.1 "once an entry from the current term is committed by counting replicas, then all
 246         // prior entries are committed indirectly".
 247         leader.handleMessage(followerActor, new AppendEntriesReply(
 248                 FOLLOWER_ID, newTerm, true, newIndex, newTerm, (short)0));
 249
 250         assertEquals("Commit Index", newIndex, actorContext.getCommitIndex());
 251     }
 252
 253     @Test
 254     public void testHandleReplicateMessageCommitIndexIncrementedBeforeConsensus() throws Exception {
 255         logStart("testHandleReplicateMessageCommitIndexIncrementedBeforeConsensus");
 256
 257         MockRaftActorContext actorContext = createActorContextWithFollower();
 258         actorContext.setRaftPolicy(createRaftPolicy(true, true));
 259
 260         long term = 1;
 261         actorContext.getTermInformation().update(term, "");
 262
 263         leader = new Leader(actorContext);
 264
 265         // Leader will send an immediate heartbeat - ignore it.
 266         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 267
 268         // The follower would normally reply - simulate that explicitly here.
 269         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 270         leader.handleMessage(followerActor, new AppendEntriesReply(
 271                 FOLLOWER_ID, term, true, lastIndex, term, (short) 0));
 272         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 273
 274         followerActor.underlyingActor().clear();
 275
 276         RaftActorBehavior raftBehavior = sendReplicate(actorContext, lastIndex + 1);
 277
 278         // State should not change
 279         assertTrue(raftBehavior instanceof Leader);
 280
 281         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 282         assertEquals("getPrevLogIndex", lastIndex, appendEntries.getPrevLogIndex());
 283         assertEquals("getPrevLogTerm", term, appendEntries.getPrevLogTerm());
 284         assertEquals("Entries size", 1, appendEntries.getEntries().size());
 285         assertEquals("Entry getIndex", lastIndex + 1, appendEntries.getEntries().get(0).getIndex());
 286         assertEquals("Entry getTerm", term, appendEntries.getEntries().get(0).getTerm());
 287         assertEquals("Entry payload", "foo", appendEntries.getEntries().get(0).getData().toString());
 288         assertEquals("Commit Index", lastIndex + 1, actorContext.getCommitIndex());
 289     }
 290
 291     @Test
 292     public void testMultipleReplicateShouldNotCauseDuplicateAppendEntriesToBeSent() throws Exception {
 293         logStart("testHandleReplicateMessageSendAppendEntriesToFollower");
 294
 295         MockRaftActorContext actorContext = createActorContextWithFollower();
 296         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
 297             @Override
 298             public FiniteDuration getHeartBeatInterval() {
 299                 return FiniteDuration.apply(5, TimeUnit.SECONDS);
 300             }
 301         });
 302
 303         long term = 1;
 304         actorContext.getTermInformation().update(term, "");
 305
 306         leader = new Leader(actorContext);
 307
 308         // Leader will send an immediate heartbeat - ignore it.
 309         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 310
 311         // The follower would normally reply - simulate that explicitly here.
 312         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 313         leader.handleMessage(followerActor, new AppendEntriesReply(
 314                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
 315         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 316
 317         followerActor.underlyingActor().clear();
 318
 319         for (int i = 0; i < 5; i++) {
 320             sendReplicate(actorContext, lastIndex + i + 1);
 321         }
 322
 323         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
 324         // We expect only 1 message to be sent because of two reasons,
 325         // - an append entries reply was not received
 326         // - the heartbeat interval has not expired
 327         // In this scenario if multiple messages are sent they would likely be duplicates
 328         assertEquals("The number of append entries collected should be 1", 1, allMessages.size());
 329     }
 330
 331     @Test
 332     public void testMultipleReplicateWithReplyShouldResultInAppendEntries() throws Exception {
 333         logStart("testMultipleReplicateWithReplyShouldResultInAppendEntries");
 334
 335         MockRaftActorContext actorContext = createActorContextWithFollower();
 336         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
 337             @Override
 338             public FiniteDuration getHeartBeatInterval() {
 339                 return FiniteDuration.apply(5, TimeUnit.SECONDS);
 340             }
 341         });
 342
 343         long term = 1;
 344         actorContext.getTermInformation().update(term, "");
 345
 346         leader = new Leader(actorContext);
 347
 348         // Leader will send an immediate heartbeat - ignore it.
 349         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 350
 351         // The follower would normally reply - simulate that explicitly here.
 352         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 353         leader.handleMessage(followerActor, new AppendEntriesReply(
 354                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
 355         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 356
 357         followerActor.underlyingActor().clear();
 358
 359         for (int i = 0; i < 3; i++) {
 360             sendReplicate(actorContext, lastIndex + i + 1);
 361             leader.handleMessage(followerActor, new AppendEntriesReply(
 362                     FOLLOWER_ID, term, true, lastIndex + i + 1, term, (short)0));
 363
 364         }
 365
 366         for (int i = 3; i < 5; i++) {
 367             sendReplicate(actorContext, lastIndex + i + 1);
 368         }
 369
 370         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
 371         // We expect 4 here because the first 3 replicate got a reply and so the 4th entry would
 372         // get sent to the follower - but not the 5th
 373         assertEquals("The number of append entries collected should be 4", 4, allMessages.size());
 374
 375         for (int i = 0; i < 4; i++) {
 376             long expected = allMessages.get(i).getEntries().get(0).getIndex();
 377             assertEquals(expected, i + 2);
 378         }
 379     }
 380
 381     @Test
 382     public void testDuplicateAppendEntriesWillBeSentOnHeartBeat() throws Exception {
 383         logStart("testDuplicateAppendEntriesWillBeSentOnHeartBeat");
 384
 385         MockRaftActorContext actorContext = createActorContextWithFollower();
 386         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
 387             @Override
 388             public FiniteDuration getHeartBeatInterval() {
 389                 return FiniteDuration.apply(500, TimeUnit.MILLISECONDS);
 390             }
 391         });
 392
 393         long term = 1;
 394         actorContext.getTermInformation().update(term, "");
 395
 396         leader = new Leader(actorContext);
 397
 398         // Leader will send an immediate heartbeat - ignore it.
 399         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 400
 401         // The follower would normally reply - simulate that explicitly here.
 402         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 403         leader.handleMessage(followerActor, new AppendEntriesReply(
 404                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
 405         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 406
 407         followerActor.underlyingActor().clear();
 408
 409         sendReplicate(actorContext, lastIndex + 1);
 410
 411         // Wait slightly longer than heartbeat duration
 412         Uninterruptibles.sleepUninterruptibly(750, TimeUnit.MILLISECONDS);
 413
 414         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
 415
 416         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
 417         assertEquals("The number of append entries collected should be 2", 2, allMessages.size());
 418
 419         assertEquals(1, allMessages.get(0).getEntries().size());
 420         assertEquals(lastIndex + 1, allMessages.get(0).getEntries().get(0).getIndex());
 421         assertEquals(1, allMessages.get(1).getEntries().size());
 422         assertEquals(lastIndex + 1, allMessages.get(0).getEntries().get(0).getIndex());
 423
 424     }
 425
 426     @Test
 427     public void testHeartbeatsAreAlwaysSentIfTheHeartbeatIntervalHasElapsed() throws Exception {
 428         logStart("testHeartbeatsAreAlwaysSentIfTheHeartbeatIntervalHasElapsed");
 429
 430         MockRaftActorContext actorContext = createActorContextWithFollower();
 431         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
 432             @Override
 433             public FiniteDuration getHeartBeatInterval() {
 434                 return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
 435             }
 436         });
 437
 438         long term = 1;
 439         actorContext.getTermInformation().update(term, "");
 440
 441         leader = new Leader(actorContext);
 442
 443         // Leader will send an immediate heartbeat - ignore it.
 444         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 445
 446         // The follower would normally reply - simulate that explicitly here.
 447         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 448         leader.handleMessage(followerActor, new AppendEntriesReply(
 449                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
 450         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 451
 452         followerActor.underlyingActor().clear();
 453
 454         for (int i = 0; i < 3; i++) {
 455             Uninterruptibles.sleepUninterruptibly(150, TimeUnit.MILLISECONDS);
 456             leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
 457         }
 458
 459         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
 460         assertEquals("The number of append entries collected should be 3", 3, allMessages.size());
 461     }
 462
 463     @Test
 464     public void testSendingReplicateImmediatelyAfterHeartbeatDoesReplicate() throws Exception {
 465         logStart("testSendingReplicateImmediatelyAfterHeartbeatDoesReplicate");
 466
 467         MockRaftActorContext actorContext = createActorContextWithFollower();
 468         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
 469             @Override
 470             public FiniteDuration getHeartBeatInterval() {
 471                 return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
 472             }
 473         });
 474
 475         long term = 1;
 476         actorContext.getTermInformation().update(term, "");
 477
 478         leader = new Leader(actorContext);
 479
 480         // Leader will send an immediate heartbeat - ignore it.
 481         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 482
 483         // The follower would normally reply - simulate that explicitly here.
 484         long lastIndex = actorContext.getReplicatedLog().lastIndex();
 485         leader.handleMessage(followerActor, new AppendEntriesReply(
 486                 FOLLOWER_ID, term, true, lastIndex, term, (short)0));
 487         assertEquals("isFollowerActive", true, leader.getFollower(FOLLOWER_ID).isFollowerActive());
 488
 489         followerActor.underlyingActor().clear();
 490
 491         Uninterruptibles.sleepUninterruptibly(150, TimeUnit.MILLISECONDS);
 492         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
 493         sendReplicate(actorContext, lastIndex + 1);
 494
 495         List<AppendEntries> allMessages = MessageCollectorActor.getAllMatching(followerActor, AppendEntries.class);
 496         assertEquals("The number of append entries collected should be 2", 2, allMessages.size());
 497
 498         assertEquals(0, allMessages.get(0).getEntries().size());
 499         assertEquals(1, allMessages.get(1).getEntries().size());
 500     }
 501
 502
 503     @Test
 504     public void testHandleReplicateMessageWhenThereAreNoFollowers() throws Exception {
 505         logStart("testHandleReplicateMessageWhenThereAreNoFollowers");
 506
 507         MockRaftActorContext actorContext = createActorContext();
 508
 509         leader = new Leader(actorContext);
 510
 511         actorContext.setLastApplied(0);
 512
 513         long newLogIndex = actorContext.getReplicatedLog().lastIndex() + 1;
 514         long term = actorContext.getTermInformation().getCurrentTerm();
 515         ReplicatedLogEntry newEntry = new SimpleReplicatedLogEntry(
 516                 newLogIndex, term, new MockRaftActorContext.MockPayload("foo"));
 517
 518         actorContext.getReplicatedLog().append(newEntry);
 519
 520         final Identifier id = new MockIdentifier("state-id");
 521         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new Replicate(leaderActor, id, newEntry));
 522
 523         // State should not change
 524         assertTrue(raftBehavior instanceof Leader);
 525
 526         assertEquals("getCommitIndex", newLogIndex, actorContext.getCommitIndex());
 527
 528         // We should get 2 ApplyState messages - 1 for new log entry and 1 for the previous
 529         // one since lastApplied state is 0.
 530         List<ApplyState> applyStateList = MessageCollectorActor.getAllMatching(
 531                 leaderActor, ApplyState.class);
 532         assertEquals("ApplyState count", newLogIndex, applyStateList.size());
 533
 534         for (int i = 0; i <= newLogIndex - 1; i++ ) {
 535             ApplyState applyState = applyStateList.get(i);
 536             assertEquals("getIndex", i + 1, applyState.getReplicatedLogEntry().getIndex());
 537             assertEquals("getTerm", term, applyState.getReplicatedLogEntry().getTerm());
 538         }
 539
 540         ApplyState last = applyStateList.get((int) newLogIndex - 1);
 541         assertEquals("getData", newEntry.getData(), last.getReplicatedLogEntry().getData());
 542         assertEquals("getIdentifier", id, last.getIdentifier());
 543     }
 544
 545     @Test
 546     public void testSendAppendEntriesOnAnInProgressInstallSnapshot() throws Exception {
 547         logStart("testSendAppendEntriesOnAnInProgressInstallSnapshot");
 548
 549         final MockRaftActorContext actorContext = createActorContextWithFollower();
 550
 551         Map<String, String> leadersSnapshot = new HashMap<>();
 552         leadersSnapshot.put("1", "A");
 553         leadersSnapshot.put("2", "B");
 554         leadersSnapshot.put("3", "C");
 555
 556         //clears leaders log
 557         actorContext.getReplicatedLog().removeFrom(0);
 558
 559         final int commitIndex = 3;
 560         final int snapshotIndex = 2;
 561         final int snapshotTerm = 1;
 562
 563         // set the snapshot variables in replicatedlog
 564         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 565         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 566         actorContext.setCommitIndex(commitIndex);
 567         //set follower timeout to 2 mins, helps during debugging
 568         actorContext.setConfigParams(new MockConfigParamsImpl(120000L, 10));
 569
 570         leader = new Leader(actorContext);
 571
 572         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
 573         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
 574
 575         //update follower timestamp
 576         leader.markFollowerActive(FOLLOWER_ID);
 577
 578         ByteString bs = toByteString(leadersSnapshot);
 579         leader.setSnapshot(Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
 580                 commitIndex, snapshotTerm, commitIndex, snapshotTerm));
 581         LeaderInstallSnapshotState fts = new LeaderInstallSnapshotState(
 582                 actorContext.getConfigParams().getSnapshotChunkSize(), leader.logName());
 583         fts.setSnapshotBytes(bs);
 584         leader.getFollower(FOLLOWER_ID).setLeaderInstallSnapshotState(fts);
 585
 586         //send first chunk and no InstallSnapshotReply received yet
 587         fts.getNextChunk();
 588         fts.incrementChunkIndex();
 589
 590         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
 591                 TimeUnit.MILLISECONDS);
 592
 593         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
 594
 595         AppendEntries ae = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 596
 597         assertTrue("AppendEntries should be sent with empty entries", ae.getEntries().isEmpty());
 598
 599         //InstallSnapshotReply received
 600         fts.markSendStatus(true);
 601
 602         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
 603
 604         InstallSnapshot is = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
 605
 606         assertEquals(commitIndex, is.getLastIncludedIndex());
 607     }
 608
 609     @Test
 610     public void testSendAppendEntriesSnapshotScenario() throws Exception {
 611         logStart("testSendAppendEntriesSnapshotScenario");
 612
 613         final MockRaftActorContext actorContext = createActorContextWithFollower();
 614
 615         Map<String, String> leadersSnapshot = new HashMap<>();
 616         leadersSnapshot.put("1", "A");
 617         leadersSnapshot.put("2", "B");
 618         leadersSnapshot.put("3", "C");
 619
 620         //clears leaders log
 621         actorContext.getReplicatedLog().removeFrom(0);
 622
 623         final int followersLastIndex = 2;
 624         final int snapshotIndex = 3;
 625         final int newEntryIndex = 4;
 626         final int snapshotTerm = 1;
 627         final int currentTerm = 2;
 628
 629         // set the snapshot variables in replicatedlog
 630         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 631         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 632         actorContext.setCommitIndex(followersLastIndex);
 633
 634         leader = new Leader(actorContext);
 635
 636         // Leader will send an immediate heartbeat - ignore it.
 637         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 638
 639         // new entry
 640         SimpleReplicatedLogEntry entry =
 641                 new SimpleReplicatedLogEntry(newEntryIndex, currentTerm,
 642                         new MockRaftActorContext.MockPayload("D"));
 643
 644         actorContext.getReplicatedLog().append(entry);
 645
 646         //update follower timestamp
 647         leader.markFollowerActive(FOLLOWER_ID);
 648
 649         // this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex
 650         RaftActorBehavior raftBehavior = leader.handleMessage(
 651                 leaderActor, new Replicate(null, new MockIdentifier("state-id"), entry));
 652
 653         assertTrue(raftBehavior instanceof Leader);
 654
 655         assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
 656     }
 657
 658     @Test
 659     public void testInitiateInstallSnapshot() throws Exception {
 660         logStart("testInitiateInstallSnapshot");
 661
 662         MockRaftActorContext actorContext = createActorContextWithFollower();
 663
 664         //clears leaders log
 665         actorContext.getReplicatedLog().removeFrom(0);
 666
 667         final int followersLastIndex = 2;
 668         final int snapshotIndex = 3;
 669         final int newEntryIndex = 4;
 670         final int snapshotTerm = 1;
 671         final int currentTerm = 2;
 672
 673         // set the snapshot variables in replicatedlog
 674         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 675         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 676         actorContext.setLastApplied(3);
 677         actorContext.setCommitIndex(followersLastIndex);
 678
 679         leader = new Leader(actorContext);
 680
 681         // Leader will send an immediate heartbeat - ignore it.
 682         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 683
 684         // set the snapshot as absent and check if capture-snapshot is invoked.
 685         leader.setSnapshot(null);
 686
 687         // new entry
 688         SimpleReplicatedLogEntry entry = new SimpleReplicatedLogEntry(newEntryIndex, currentTerm,
 689                 new MockRaftActorContext.MockPayload("D"));
 690
 691         actorContext.getReplicatedLog().append(entry);
 692
 693         //update follower timestamp
 694         leader.markFollowerActive(FOLLOWER_ID);
 695
 696         leader.handleMessage(leaderActor, new Replicate(null, new MockIdentifier("state-id"), entry));
 697
 698         assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
 699
 700         CaptureSnapshot cs = actorContext.getSnapshotManager().getCaptureSnapshot();
 701
 702         assertTrue(cs.isInstallSnapshotInitiated());
 703         assertEquals(3, cs.getLastAppliedIndex());
 704         assertEquals(1, cs.getLastAppliedTerm());
 705         assertEquals(4, cs.getLastIndex());
 706         assertEquals(2, cs.getLastTerm());
 707
 708         // if an initiate is started again when first is in progress, it shouldnt initiate Capture
 709         leader.handleMessage(leaderActor, new Replicate(null, new MockIdentifier("state-id"), entry));
 710
 711         assertSame("CaptureSnapshot instance", cs, actorContext.getSnapshotManager().getCaptureSnapshot());
 712     }
 713
 714     @Test
 715     public void testInitiateForceInstallSnapshot() throws Exception {
 716         logStart("testInitiateForceInstallSnapshot");
 717
 718         MockRaftActorContext actorContext = createActorContextWithFollower();
 719
 720         final int followersLastIndex = 2;
 721         final int snapshotIndex = -1;
 722         final int newEntryIndex = 4;
 723         final int snapshotTerm = -1;
 724         final int currentTerm = 2;
 725
 726         // set the snapshot variables in replicatedlog
 727         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 728         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 729         actorContext.setLastApplied(3);
 730         actorContext.setCommitIndex(followersLastIndex);
 731
 732         actorContext.getReplicatedLog().removeFrom(0);
 733
 734         leader = new Leader(actorContext);
 735         actorContext.setCurrentBehavior(leader);
 736
 737         // Leader will send an immediate heartbeat - ignore it.
 738         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 739
 740         // set the snapshot as absent and check if capture-snapshot is invoked.
 741         leader.setSnapshot(null);
 742
 743         for (int i = 0; i < 4; i++) {
 744             actorContext.getReplicatedLog().append(new SimpleReplicatedLogEntry(i, 1,
 745                     new MockRaftActorContext.MockPayload("X" + i)));
 746         }
 747
 748         // new entry
 749         SimpleReplicatedLogEntry entry = new SimpleReplicatedLogEntry(newEntryIndex, currentTerm,
 750                 new MockRaftActorContext.MockPayload("D"));
 751
 752         actorContext.getReplicatedLog().append(entry);
 753
 754         //update follower timestamp
 755         leader.markFollowerActive(FOLLOWER_ID);
 756
 757         // Sending this AppendEntriesReply forces the Leader to capture a snapshot, which subsequently gets
 758         // installed with a SendInstallSnapshot
 759         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, false, 1, 1, (short) 1, true));
 760
 761         assertEquals("isCapturing", true, actorContext.getSnapshotManager().isCapturing());
 762
 763         CaptureSnapshot cs = actorContext.getSnapshotManager().getCaptureSnapshot();
 764
 765         assertTrue(cs.isInstallSnapshotInitiated());
 766         assertEquals(3, cs.getLastAppliedIndex());
 767         assertEquals(1, cs.getLastAppliedTerm());
 768         assertEquals(4, cs.getLastIndex());
 769         assertEquals(2, cs.getLastTerm());
 770
 771         // if an initiate is started again when first is in progress, it should not initiate Capture
 772         leader.handleMessage(leaderActor, new Replicate(null, new MockIdentifier("state-id"), entry));
 773
 774         assertSame("CaptureSnapshot instance", cs, actorContext.getSnapshotManager().getCaptureSnapshot());
 775     }
 776
 777
 778     @Test
 779     public void testInstallSnapshot() throws Exception {
 780         logStart("testInstallSnapshot");
 781
 782         final MockRaftActorContext actorContext = createActorContextWithFollower();
 783
 784         Map<String, String> leadersSnapshot = new HashMap<>();
 785         leadersSnapshot.put("1", "A");
 786         leadersSnapshot.put("2", "B");
 787         leadersSnapshot.put("3", "C");
 788
 789         //clears leaders log
 790         actorContext.getReplicatedLog().removeFrom(0);
 791
 792         final int lastAppliedIndex = 3;
 793         final int snapshotIndex = 2;
 794         final int snapshotTerm = 1;
 795         final int currentTerm = 2;
 796
 797         // set the snapshot variables in replicatedlog
 798         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 799         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 800         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
 801         actorContext.setCommitIndex(lastAppliedIndex);
 802         actorContext.setLastApplied(lastAppliedIndex);
 803
 804         leader = new Leader(actorContext);
 805
 806         // Initial heartbeat.
 807         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 808
 809         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
 810         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
 811
 812         Snapshot snapshot = Snapshot.create(toByteString(leadersSnapshot).toByteArray(),
 813                 Collections.<ReplicatedLogEntry>emptyList(),
 814                 lastAppliedIndex, snapshotTerm, lastAppliedIndex, snapshotTerm);
 815
 816         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
 817
 818         assertTrue(raftBehavior instanceof Leader);
 819
 820         // check if installsnapshot gets called with the correct values.
 821
 822         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
 823                 InstallSnapshot.class);
 824
 825         assertNotNull(installSnapshot.getData());
 826         assertEquals(lastAppliedIndex, installSnapshot.getLastIncludedIndex());
 827         assertEquals(snapshotTerm, installSnapshot.getLastIncludedTerm());
 828
 829         assertEquals(currentTerm, installSnapshot.getTerm());
 830     }
 831
 832     @Test
 833     public void testForceInstallSnapshot() throws Exception {
 834         logStart("testForceInstallSnapshot");
 835
 836         final MockRaftActorContext actorContext = createActorContextWithFollower();
 837
 838         Map<String, String> leadersSnapshot = new HashMap<>();
 839         leadersSnapshot.put("1", "A");
 840         leadersSnapshot.put("2", "B");
 841         leadersSnapshot.put("3", "C");
 842
 843         final int lastAppliedIndex = 3;
 844         final int snapshotIndex = -1;
 845         final int snapshotTerm = -1;
 846         final int currentTerm = 2;
 847
 848         // set the snapshot variables in replicatedlog
 849         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 850         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 851         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
 852         actorContext.setCommitIndex(lastAppliedIndex);
 853         actorContext.setLastApplied(lastAppliedIndex);
 854
 855         leader = new Leader(actorContext);
 856
 857         // Initial heartbeat.
 858         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 859
 860         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
 861         leader.getFollower(FOLLOWER_ID).setNextIndex(-1);
 862
 863         Snapshot snapshot = Snapshot.create(toByteString(leadersSnapshot).toByteArray(),
 864                 Collections.<ReplicatedLogEntry>emptyList(),
 865                 lastAppliedIndex, snapshotTerm, lastAppliedIndex, snapshotTerm);
 866
 867         RaftActorBehavior raftBehavior = leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
 868
 869         assertTrue(raftBehavior instanceof Leader);
 870
 871         // check if installsnapshot gets called with the correct values.
 872
 873         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
 874                 InstallSnapshot.class);
 875
 876         assertNotNull(installSnapshot.getData());
 877         assertEquals(lastAppliedIndex, installSnapshot.getLastIncludedIndex());
 878         assertEquals(snapshotTerm, installSnapshot.getLastIncludedTerm());
 879
 880         assertEquals(currentTerm, installSnapshot.getTerm());
 881     }
 882
 883     @Test
 884     public void testHandleInstallSnapshotReplyLastChunk() throws Exception {
 885         logStart("testHandleInstallSnapshotReplyLastChunk");
 886
 887         MockRaftActorContext actorContext = createActorContextWithFollower();
 888
 889         final int commitIndex = 3;
 890         final int snapshotIndex = 2;
 891         final int snapshotTerm = 1;
 892         final int currentTerm = 2;
 893
 894         actorContext.setCommitIndex(commitIndex);
 895
 896         leader = new Leader(actorContext);
 897         actorContext.setCurrentBehavior(leader);
 898
 899         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
 900         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
 901
 902         // Ignore initial heartbeat.
 903         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
 904
 905         Map<String, String> leadersSnapshot = new HashMap<>();
 906         leadersSnapshot.put("1", "A");
 907         leadersSnapshot.put("2", "B");
 908         leadersSnapshot.put("3", "C");
 909
 910         // set the snapshot variables in replicatedlog
 911
 912         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 913         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 914         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
 915
 916         ByteString bs = toByteString(leadersSnapshot);
 917         leader.setSnapshot(Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
 918                 commitIndex, snapshotTerm, commitIndex, snapshotTerm));
 919         LeaderInstallSnapshotState fts = new LeaderInstallSnapshotState(
 920                 actorContext.getConfigParams().getSnapshotChunkSize(), leader.logName());
 921         fts.setSnapshotBytes(bs);
 922         leader.getFollower(FOLLOWER_ID).setLeaderInstallSnapshotState(fts);
 923         while (!fts.isLastChunk(fts.getChunkIndex())) {
 924             fts.getNextChunk();
 925             fts.incrementChunkIndex();
 926         }
 927
 928         //clears leaders log
 929         actorContext.getReplicatedLog().removeFrom(0);
 930
 931         RaftActorBehavior raftBehavior = leader.handleMessage(followerActor,
 932                 new InstallSnapshotReply(currentTerm, FOLLOWER_ID, fts.getChunkIndex(), true));
 933
 934         assertTrue(raftBehavior instanceof Leader);
 935
 936         assertEquals(1, leader.followerLogSize());
 937         FollowerLogInformation fli = leader.getFollower(FOLLOWER_ID);
 938         assertNotNull(fli);
 939         assertNull(fli.getInstallSnapshotState());
 940         assertEquals(commitIndex, fli.getMatchIndex());
 941         assertEquals(commitIndex + 1, fli.getNextIndex());
 942         assertFalse(leader.hasSnapshot());
 943     }
 944
 945     @Test
 946     public void testSendSnapshotfromInstallSnapshotReply() throws Exception {
 947         logStart("testSendSnapshotfromInstallSnapshotReply");
 948
 949         MockRaftActorContext actorContext = createActorContextWithFollower();
 950
 951         final int commitIndex = 3;
 952         final int snapshotIndex = 2;
 953         final int snapshotTerm = 1;
 954         final int currentTerm = 2;
 955
 956         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl() {
 957             @Override
 958             public int getSnapshotChunkSize() {
 959                 return 50;
 960             }
 961         };
 962         configParams.setHeartBeatInterval(new FiniteDuration(9, TimeUnit.SECONDS));
 963         configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
 964
 965         actorContext.setConfigParams(configParams);
 966         actorContext.setCommitIndex(commitIndex);
 967
 968         leader = new Leader(actorContext);
 969         actorContext.setCurrentBehavior(leader);
 970
 971         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
 972         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
 973
 974         Map<String, String> leadersSnapshot = new HashMap<>();
 975         leadersSnapshot.put("1", "A");
 976         leadersSnapshot.put("2", "B");
 977         leadersSnapshot.put("3", "C");
 978
 979         // set the snapshot variables in replicatedlog
 980         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
 981         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
 982         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
 983
 984         ByteString bs = toByteString(leadersSnapshot);
 985         Snapshot snapshot = Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
 986                 commitIndex, snapshotTerm, commitIndex, snapshotTerm);
 987         leader.setSnapshot(snapshot);
 988
 989         leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
 990
 991         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
 992                 InstallSnapshot.class);
 993
 994         assertEquals(1, installSnapshot.getChunkIndex());
 995         assertEquals(3, installSnapshot.getTotalChunks());
 996
 997         followerActor.underlyingActor().clear();
 998         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
 999                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
1000
1001         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1002
1003         assertEquals(2, installSnapshot.getChunkIndex());
1004         assertEquals(3, installSnapshot.getTotalChunks());
1005
1006         followerActor.underlyingActor().clear();
1007         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
1008                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
1009
1010         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1011
1012         // Send snapshot reply one more time and make sure that a new snapshot message should not be sent to follower
1013         followerActor.underlyingActor().clear();
1014         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
1015                 FOLLOWER_ID, installSnapshot.getChunkIndex(), true));
1016
1017         installSnapshot = MessageCollectorActor.getFirstMatching(followerActor, InstallSnapshot.class);
1018
1019         assertNull(installSnapshot);
1020     }
1021
1022
1023     @Test
1024     public void testHandleInstallSnapshotReplyWithInvalidChunkIndex() throws Exception {
1025         logStart("testHandleInstallSnapshotReplyWithInvalidChunkIndex");
1026
1027         MockRaftActorContext actorContext = createActorContextWithFollower();
1028
1029         final int commitIndex = 3;
1030         final int snapshotIndex = 2;
1031         final int snapshotTerm = 1;
1032         final int currentTerm = 2;
1033
1034         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
1035             @Override
1036             public int getSnapshotChunkSize() {
1037                 return 50;
1038             }
1039         });
1040
1041         actorContext.setCommitIndex(commitIndex);
1042
1043         leader = new Leader(actorContext);
1044
1045         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
1046         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
1047
1048         Map<String, String> leadersSnapshot = new HashMap<>();
1049         leadersSnapshot.put("1", "A");
1050         leadersSnapshot.put("2", "B");
1051         leadersSnapshot.put("3", "C");
1052
1053         // set the snapshot variables in replicatedlog
1054         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
1055         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
1056         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
1057
1058         ByteString bs = toByteString(leadersSnapshot);
1059         Snapshot snapshot = Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
1060                 commitIndex, snapshotTerm, commitIndex, snapshotTerm);
1061         leader.setSnapshot(snapshot);
1062
1063         Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS);
1064         leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
1065
1066         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
1067                 InstallSnapshot.class);
1068
1069         assertEquals(1, installSnapshot.getChunkIndex());
1070         assertEquals(3, installSnapshot.getTotalChunks());
1071
1072         followerActor.underlyingActor().clear();
1073
1074         leader.handleMessage(followerActor, new InstallSnapshotReply(actorContext.getTermInformation().getCurrentTerm(),
1075                 FOLLOWER_ID, -1, false));
1076
1077         Uninterruptibles.sleepUninterruptibly(actorContext.getConfigParams().getHeartBeatInterval().toMillis(),
1078                 TimeUnit.MILLISECONDS);
1079
1080         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
1081
1082         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1083
1084         assertEquals(1, installSnapshot.getChunkIndex());
1085         assertEquals(3, installSnapshot.getTotalChunks());
1086     }
1087
1088     @Test
1089     public void testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk() throws Exception {
1090         logStart("testHandleSnapshotSendsPreviousChunksHashCodeWhenSendingNextChunk");
1091
1092         MockRaftActorContext actorContext = createActorContextWithFollower();
1093
1094         final int commitIndex = 3;
1095         final int snapshotIndex = 2;
1096         final int snapshotTerm = 1;
1097         final int currentTerm = 2;
1098
1099         actorContext.setConfigParams(new DefaultConfigParamsImpl() {
1100             @Override
1101             public int getSnapshotChunkSize() {
1102                 return 50;
1103             }
1104         });
1105
1106         actorContext.setCommitIndex(commitIndex);
1107
1108         leader = new Leader(actorContext);
1109
1110         leader.getFollower(FOLLOWER_ID).setMatchIndex(-1);
1111         leader.getFollower(FOLLOWER_ID).setNextIndex(0);
1112
1113         Map<String, String> leadersSnapshot = new HashMap<>();
1114         leadersSnapshot.put("1", "A");
1115         leadersSnapshot.put("2", "B");
1116         leadersSnapshot.put("3", "C");
1117
1118         // set the snapshot variables in replicatedlog
1119         actorContext.getReplicatedLog().setSnapshotIndex(snapshotIndex);
1120         actorContext.getReplicatedLog().setSnapshotTerm(snapshotTerm);
1121         actorContext.getTermInformation().update(currentTerm, leaderActor.path().toString());
1122
1123         ByteString bs = toByteString(leadersSnapshot);
1124         Snapshot snapshot = Snapshot.create(bs.toByteArray(), Collections.<ReplicatedLogEntry>emptyList(),
1125                 commitIndex, snapshotTerm, commitIndex, snapshotTerm);
1126         leader.setSnapshot(snapshot);
1127
1128         leader.handleMessage(leaderActor, new SendInstallSnapshot(snapshot));
1129
1130         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
1131                 InstallSnapshot.class);
1132
1133         assertEquals(1, installSnapshot.getChunkIndex());
1134         assertEquals(3, installSnapshot.getTotalChunks());
1135         assertEquals(LeaderInstallSnapshotState.INITIAL_LAST_CHUNK_HASH_CODE,
1136                 installSnapshot.getLastChunkHashCode().get().intValue());
1137
1138         final int hashCode = Arrays.hashCode(installSnapshot.getData());
1139
1140         followerActor.underlyingActor().clear();
1141
1142         leader.handleMessage(followerActor, new InstallSnapshotReply(installSnapshot.getTerm(),
1143                 FOLLOWER_ID, 1, true));
1144
1145         installSnapshot = MessageCollectorActor.expectFirstMatching(followerActor, InstallSnapshot.class);
1146
1147         assertEquals(2, installSnapshot.getChunkIndex());
1148         assertEquals(3, installSnapshot.getTotalChunks());
1149         assertEquals(hashCode, installSnapshot.getLastChunkHashCode().get().intValue());
1150     }
1151
1152     @Test
1153     public void testLeaderInstallSnapshotState() {
1154         logStart("testLeaderInstallSnapshotState");
1155
1156         Map<String, String> leadersSnapshot = new HashMap<>();
1157         leadersSnapshot.put("1", "A");
1158         leadersSnapshot.put("2", "B");
1159         leadersSnapshot.put("3", "C");
1160
1161         ByteString bs = toByteString(leadersSnapshot);
1162         byte[] barray = bs.toByteArray();
1163
1164         LeaderInstallSnapshotState fts = new LeaderInstallSnapshotState(50, "test");
1165         fts.setSnapshotBytes(bs);
1166
1167         assertEquals(bs.size(), barray.length);
1168
1169         int chunkIndex = 0;
1170         for (int i = 0; i < barray.length; i = i + 50) {
1171             int length = i + 50;
1172             chunkIndex++;
1173
1174             if (i + 50 > barray.length) {
1175                 length = barray.length;
1176             }
1177
1178             byte[] chunk = fts.getNextChunk();
1179             assertEquals("bytestring size not matching for chunk:" + chunkIndex, length - i, chunk.length);
1180             assertEquals("chunkindex not matching", chunkIndex, fts.getChunkIndex());
1181
1182             fts.markSendStatus(true);
1183             if (!fts.isLastChunk(chunkIndex)) {
1184                 fts.incrementChunkIndex();
1185             }
1186         }
1187
1188         assertEquals("totalChunks not matching", chunkIndex, fts.getTotalChunks());
1189     }
1190
1191     @Override
1192     protected Leader createBehavior(final RaftActorContext actorContext) {
1193         return new Leader(actorContext);
1194     }
1195
1196     @Override
1197     protected MockRaftActorContext createActorContext() {
1198         return createActorContext(leaderActor);
1199     }
1200
1201     @Override
1202     protected MockRaftActorContext createActorContext(ActorRef actorRef) {
1203         return createActorContext(LEADER_ID, actorRef);
1204     }
1205
1206     private MockRaftActorContext createActorContext(String id, ActorRef actorRef) {
1207         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
1208         configParams.setHeartBeatInterval(new FiniteDuration(50, TimeUnit.MILLISECONDS));
1209         configParams.setElectionTimeoutFactor(100000);
1210         MockRaftActorContext context = new MockRaftActorContext(id, getSystem(), actorRef);
1211         context.setConfigParams(configParams);
1212         context.setPayloadVersion(payloadVersion);
1213         return context;
1214     }
1215
1216     private MockRaftActorContext createActorContextWithFollower() {
1217         MockRaftActorContext actorContext = createActorContext();
1218         actorContext.setPeerAddresses(ImmutableMap.<String, String>builder().put(FOLLOWER_ID,
1219                 followerActor.path().toString()).build());
1220         return actorContext;
1221     }
1222
1223     private MockRaftActorContext createFollowerActorContextWithLeader() {
1224         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
1225         DefaultConfigParamsImpl followerConfig = new DefaultConfigParamsImpl();
1226         followerConfig.setElectionTimeoutFactor(10000);
1227         followerActorContext.setConfigParams(followerConfig);
1228         followerActorContext.setPeerAddresses(ImmutableMap.of(LEADER_ID, leaderActor.path().toString()));
1229         return followerActorContext;
1230     }
1231
1232     @Test
1233     public void testLeaderCreatedWithCommitIndexLessThanLastIndex() throws Exception {
1234         logStart("testLeaderCreatedWithCommitIndexLessThanLastIndex");
1235
1236         final MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1237
1238         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
1239
1240         Follower follower = new Follower(followerActorContext);
1241         followerActor.underlyingActor().setBehavior(follower);
1242         followerActorContext.setCurrentBehavior(follower);
1243
1244         Map<String, String> peerAddresses = new HashMap<>();
1245         peerAddresses.put(FOLLOWER_ID, followerActor.path().toString());
1246
1247         leaderActorContext.setPeerAddresses(peerAddresses);
1248
1249         leaderActorContext.getReplicatedLog().removeFrom(0);
1250
1251         //create 3 entries
1252         leaderActorContext.setReplicatedLog(
1253                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1254
1255         leaderActorContext.setCommitIndex(1);
1256
1257         followerActorContext.getReplicatedLog().removeFrom(0);
1258
1259         // follower too has the exact same log entries and has the same commit index
1260         followerActorContext.setReplicatedLog(
1261                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1262
1263         followerActorContext.setCommitIndex(1);
1264
1265         leader = new Leader(leaderActorContext);
1266         leaderActorContext.setCurrentBehavior(leader);
1267
1268         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1269
1270         assertEquals(-1, appendEntries.getLeaderCommit());
1271         assertEquals(0, appendEntries.getEntries().size());
1272         assertEquals(0, appendEntries.getPrevLogIndex());
1273
1274         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(
1275                 leaderActor, AppendEntriesReply.class);
1276
1277         assertEquals(2, appendEntriesReply.getLogLastIndex());
1278         assertEquals(1, appendEntriesReply.getLogLastTerm());
1279
1280         // follower returns its next index
1281         assertEquals(2, appendEntriesReply.getLogLastIndex());
1282         assertEquals(1, appendEntriesReply.getLogLastTerm());
1283
1284         follower.close();
1285     }
1286
1287     @Test
1288     public void testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex() throws Exception {
1289         logStart("testLeaderCreatedWithCommitIndexLessThanFollowersCommitIndex");
1290
1291         final MockRaftActorContext leaderActorContext = createActorContext();
1292
1293         MockRaftActorContext followerActorContext = createActorContext(FOLLOWER_ID, followerActor);
1294         followerActorContext.setPeerAddresses(ImmutableMap.of(LEADER_ID, leaderActor.path().toString()));
1295
1296         Follower follower = new Follower(followerActorContext);
1297         followerActor.underlyingActor().setBehavior(follower);
1298         followerActorContext.setCurrentBehavior(follower);
1299
1300         Map<String, String> leaderPeerAddresses = new HashMap<>();
1301         leaderPeerAddresses.put(FOLLOWER_ID, followerActor.path().toString());
1302
1303         leaderActorContext.setPeerAddresses(leaderPeerAddresses);
1304
1305         leaderActorContext.getReplicatedLog().removeFrom(0);
1306
1307         leaderActorContext.setReplicatedLog(
1308                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1309
1310         leaderActorContext.setCommitIndex(1);
1311
1312         followerActorContext.getReplicatedLog().removeFrom(0);
1313
1314         followerActorContext.setReplicatedLog(
1315                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1316
1317         // follower has the same log entries but its commit index > leaders commit index
1318         followerActorContext.setCommitIndex(2);
1319
1320         leader = new Leader(leaderActorContext);
1321
1322         // Initial heartbeat
1323         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1324
1325         assertEquals(-1, appendEntries.getLeaderCommit());
1326         assertEquals(0, appendEntries.getEntries().size());
1327         assertEquals(0, appendEntries.getPrevLogIndex());
1328
1329         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(
1330                 leaderActor, AppendEntriesReply.class);
1331
1332         assertEquals(2, appendEntriesReply.getLogLastIndex());
1333         assertEquals(1, appendEntriesReply.getLogLastTerm());
1334
1335         leaderActor.underlyingActor().setBehavior(follower);
1336         leader.handleMessage(followerActor, appendEntriesReply);
1337
1338         leaderActor.underlyingActor().clear();
1339         followerActor.underlyingActor().clear();
1340
1341         Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams().getHeartBeatInterval().toMillis(),
1342                 TimeUnit.MILLISECONDS);
1343
1344         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
1345
1346         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1347
1348         assertEquals(2, appendEntries.getLeaderCommit());
1349         assertEquals(0, appendEntries.getEntries().size());
1350         assertEquals(2, appendEntries.getPrevLogIndex());
1351
1352         appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
1353
1354         assertEquals(2, appendEntriesReply.getLogLastIndex());
1355         assertEquals(1, appendEntriesReply.getLogLastTerm());
1356
1357         assertEquals(2, followerActorContext.getCommitIndex());
1358
1359         follower.close();
1360     }
1361
1362     @Test
1363     public void testHandleAppendEntriesReplyFailureWithFollowersLogBehindTheLeader() {
1364         logStart("testHandleAppendEntriesReplyFailureWithFollowersLogBehindTheLeader");
1365
1366         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1367         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1368                 new FiniteDuration(1000, TimeUnit.SECONDS));
1369
1370         leaderActorContext.setReplicatedLog(
1371                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1372         long leaderCommitIndex = 2;
1373         leaderActorContext.setCommitIndex(leaderCommitIndex);
1374         leaderActorContext.setLastApplied(leaderCommitIndex);
1375
1376         final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1377         final ReplicatedLogEntry leadersThirdLogEntry = leaderActorContext.getReplicatedLog().get(2);
1378
1379         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1380
1381         followerActorContext.setReplicatedLog(
1382                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build());
1383         followerActorContext.setCommitIndex(0);
1384         followerActorContext.setLastApplied(0);
1385
1386         Follower follower = new Follower(followerActorContext);
1387         followerActor.underlyingActor().setBehavior(follower);
1388
1389         leader = new Leader(leaderActorContext);
1390
1391         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1392         final AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor,
1393                 AppendEntriesReply.class);
1394
1395         MessageCollectorActor.clearMessages(followerActor);
1396         MessageCollectorActor.clearMessages(leaderActor);
1397
1398         // Verify initial AppendEntries sent.
1399         assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit());
1400         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1401         assertEquals("getPrevLogIndex", 1, appendEntries.getPrevLogIndex());
1402
1403         leaderActor.underlyingActor().setBehavior(leader);
1404
1405         leader.handleMessage(followerActor, appendEntriesReply);
1406
1407         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1);
1408         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1409
1410         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1411         assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex());
1412         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1413
1414         assertEquals("First entry index", 1, appendEntries.getEntries().get(0).getIndex());
1415         assertEquals("First entry data", leadersSecondLogEntry.getData(),
1416                 appendEntries.getEntries().get(0).getData());
1417         assertEquals("Second entry index", 2, appendEntries.getEntries().get(1).getIndex());
1418         assertEquals("Second entry data", leadersThirdLogEntry.getData(),
1419                 appendEntries.getEntries().get(1).getData());
1420
1421         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1422         assertEquals("getNextIndex", 3, followerInfo.getNextIndex());
1423
1424         List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2);
1425
1426         ApplyState applyState = applyStateList.get(0);
1427         assertEquals("Follower's first ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex());
1428         assertEquals("Follower's first ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1429         assertEquals("Follower's first ApplyState data", leadersSecondLogEntry.getData(),
1430                 applyState.getReplicatedLogEntry().getData());
1431
1432         applyState = applyStateList.get(1);
1433         assertEquals("Follower's second ApplyState index", 2, applyState.getReplicatedLogEntry().getIndex());
1434         assertEquals("Follower's second ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1435         assertEquals("Follower's second ApplyState data", leadersThirdLogEntry.getData(),
1436                 applyState.getReplicatedLogEntry().getData());
1437
1438         assertEquals("Follower's commit index", 2, followerActorContext.getCommitIndex());
1439         assertEquals("Follower's lastIndex", 2, followerActorContext.getReplicatedLog().lastIndex());
1440     }
1441
1442     @Test
1443     public void testHandleAppendEntriesReplyFailureWithFollowersLogEmpty() {
1444         logStart("testHandleAppendEntriesReplyFailureWithFollowersLogEmpty");
1445
1446         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1447         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1448                 new FiniteDuration(1000, TimeUnit.SECONDS));
1449
1450         leaderActorContext.setReplicatedLog(
1451                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build());
1452         long leaderCommitIndex = 1;
1453         leaderActorContext.setCommitIndex(leaderCommitIndex);
1454         leaderActorContext.setLastApplied(leaderCommitIndex);
1455
1456         final ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0);
1457         final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1458
1459         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1460
1461         followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
1462         followerActorContext.setCommitIndex(-1);
1463         followerActorContext.setLastApplied(-1);
1464
1465         Follower follower = new Follower(followerActorContext);
1466         followerActor.underlyingActor().setBehavior(follower);
1467         followerActorContext.setCurrentBehavior(follower);
1468
1469         leader = new Leader(leaderActorContext);
1470
1471         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1472         final AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor,
1473                 AppendEntriesReply.class);
1474
1475         MessageCollectorActor.clearMessages(followerActor);
1476         MessageCollectorActor.clearMessages(leaderActor);
1477
1478         // Verify initial AppendEntries sent with the leader's current commit index.
1479         assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit());
1480         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1481         assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex());
1482
1483         leaderActor.underlyingActor().setBehavior(leader);
1484         leaderActorContext.setCurrentBehavior(leader);
1485
1486         leader.handleMessage(followerActor, appendEntriesReply);
1487
1488         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1);
1489         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1490
1491         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1492         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1493         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1494
1495         assertEquals("First entry index", 0, appendEntries.getEntries().get(0).getIndex());
1496         assertEquals("First entry data", leadersFirstLogEntry.getData(),
1497                 appendEntries.getEntries().get(0).getData());
1498         assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex());
1499         assertEquals("Second entry data", leadersSecondLogEntry.getData(),
1500                 appendEntries.getEntries().get(1).getData());
1501
1502         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1503         assertEquals("getNextIndex", 2, followerInfo.getNextIndex());
1504
1505         List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2);
1506
1507         ApplyState applyState = applyStateList.get(0);
1508         assertEquals("Follower's first ApplyState index", 0, applyState.getReplicatedLogEntry().getIndex());
1509         assertEquals("Follower's first ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1510         assertEquals("Follower's first ApplyState data", leadersFirstLogEntry.getData(),
1511                 applyState.getReplicatedLogEntry().getData());
1512
1513         applyState = applyStateList.get(1);
1514         assertEquals("Follower's second ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex());
1515         assertEquals("Follower's second ApplyState term", 1, applyState.getReplicatedLogEntry().getTerm());
1516         assertEquals("Follower's second ApplyState data", leadersSecondLogEntry.getData(),
1517                 applyState.getReplicatedLogEntry().getData());
1518
1519         assertEquals("Follower's commit index", 1, followerActorContext.getCommitIndex());
1520         assertEquals("Follower's lastIndex", 1, followerActorContext.getReplicatedLog().lastIndex());
1521     }
1522
1523     @Test
1524     public void testHandleAppendEntriesReplyFailureWithFollowersLogTermDifferent() {
1525         logStart("testHandleAppendEntriesReplyFailureWithFollowersLogTermDifferent");
1526
1527         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1528         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1529                 new FiniteDuration(1000, TimeUnit.SECONDS));
1530
1531         leaderActorContext.setReplicatedLog(
1532                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 2).build());
1533         long leaderCommitIndex = 1;
1534         leaderActorContext.setCommitIndex(leaderCommitIndex);
1535         leaderActorContext.setLastApplied(leaderCommitIndex);
1536
1537         final ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0);
1538         final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1539
1540         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1541
1542         followerActorContext.setReplicatedLog(
1543                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build());
1544         followerActorContext.setCommitIndex(-1);
1545         followerActorContext.setLastApplied(-1);
1546
1547         Follower follower = new Follower(followerActorContext);
1548         followerActor.underlyingActor().setBehavior(follower);
1549         followerActorContext.setCurrentBehavior(follower);
1550
1551         leader = new Leader(leaderActorContext);
1552
1553         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1554         final AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor,
1555                 AppendEntriesReply.class);
1556
1557         MessageCollectorActor.clearMessages(followerActor);
1558         MessageCollectorActor.clearMessages(leaderActor);
1559
1560         // Verify initial AppendEntries sent with the leader's current commit index.
1561         assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit());
1562         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1563         assertEquals("getPrevLogIndex", 0, appendEntries.getPrevLogIndex());
1564
1565         leaderActor.underlyingActor().setBehavior(leader);
1566         leaderActorContext.setCurrentBehavior(leader);
1567
1568         leader.handleMessage(followerActor, appendEntriesReply);
1569
1570         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 1);
1571         appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1572
1573         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1574         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1575         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1576
1577         assertEquals("First entry index", 0, appendEntries.getEntries().get(0).getIndex());
1578         assertEquals("First entry term", 2, appendEntries.getEntries().get(0).getTerm());
1579         assertEquals("First entry data", leadersFirstLogEntry.getData(),
1580                 appendEntries.getEntries().get(0).getData());
1581         assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex());
1582         assertEquals("Second entry term", 2, appendEntries.getEntries().get(1).getTerm());
1583         assertEquals("Second entry data", leadersSecondLogEntry.getData(),
1584                 appendEntries.getEntries().get(1).getData());
1585
1586         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1587         assertEquals("getNextIndex", 2, followerInfo.getNextIndex());
1588
1589         List<ApplyState> applyStateList = MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 2);
1590
1591         ApplyState applyState = applyStateList.get(0);
1592         assertEquals("Follower's first ApplyState index", 0, applyState.getReplicatedLogEntry().getIndex());
1593         assertEquals("Follower's first ApplyState term", 2, applyState.getReplicatedLogEntry().getTerm());
1594         assertEquals("Follower's first ApplyState data", leadersFirstLogEntry.getData(),
1595                 applyState.getReplicatedLogEntry().getData());
1596
1597         applyState = applyStateList.get(1);
1598         assertEquals("Follower's second ApplyState index", 1, applyState.getReplicatedLogEntry().getIndex());
1599         assertEquals("Follower's second ApplyState term", 2, applyState.getReplicatedLogEntry().getTerm());
1600         assertEquals("Follower's second ApplyState data", leadersSecondLogEntry.getData(),
1601                 applyState.getReplicatedLogEntry().getData());
1602
1603         assertEquals("Follower's commit index", 1, followerActorContext.getCommitIndex());
1604         assertEquals("Follower's lastIndex", 1, followerActorContext.getReplicatedLog().lastIndex());
1605         assertEquals("Follower's lastTerm", 2, followerActorContext.getReplicatedLog().lastTerm());
1606     }
1607
1608     @Test
1609     public void testHandleAppendEntriesReplyWithNewerTerm() {
1610         logStart("testHandleAppendEntriesReplyWithNewerTerm");
1611
1612         MockRaftActorContext leaderActorContext = createActorContext();
1613         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1614                 new FiniteDuration(10000, TimeUnit.SECONDS));
1615
1616         leaderActorContext.setReplicatedLog(
1617                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 2).build());
1618
1619         leader = new Leader(leaderActorContext);
1620         leaderActor.underlyingActor().setBehavior(leader);
1621         leaderActor.tell(new AppendEntriesReply("foo", 20, false, 1000, 10, (short) 1), ActorRef.noSender());
1622
1623         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor,
1624                 AppendEntriesReply.class);
1625
1626         assertEquals(false, appendEntriesReply.isSuccess());
1627         assertEquals(RaftState.Follower, leaderActor.underlyingActor().getFirstBehaviorChange().state());
1628
1629         MessageCollectorActor.clearMessages(leaderActor);
1630     }
1631
1632     @Test
1633     public void testHandleAppendEntriesReplyWithNewerTermWhenElectionsAreDisabled() {
1634         logStart("testHandleAppendEntriesReplyWithNewerTermWhenElectionsAreDisabled");
1635
1636         MockRaftActorContext leaderActorContext = createActorContext();
1637         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1638                 new FiniteDuration(10000, TimeUnit.SECONDS));
1639
1640         leaderActorContext.setReplicatedLog(
1641                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 2).build());
1642         leaderActorContext.setRaftPolicy(createRaftPolicy(false, false));
1643
1644         leader = new Leader(leaderActorContext);
1645         leaderActor.underlyingActor().setBehavior(leader);
1646         leaderActor.tell(new AppendEntriesReply("foo", 20, false, 1000, 10, (short) 1), ActorRef.noSender());
1647
1648         AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor,
1649                 AppendEntriesReply.class);
1650
1651         assertEquals(false, appendEntriesReply.isSuccess());
1652         assertEquals(RaftState.Leader, leaderActor.underlyingActor().getFirstBehaviorChange().state());
1653
1654         MessageCollectorActor.clearMessages(leaderActor);
1655     }
1656
1657     @Test
1658     public void testHandleAppendEntriesReplySuccess() throws Exception {
1659         logStart("testHandleAppendEntriesReplySuccess");
1660
1661         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1662
1663         leaderActorContext.setReplicatedLog(
1664                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build());
1665
1666         leaderActorContext.setCommitIndex(1);
1667         leaderActorContext.setLastApplied(1);
1668         leaderActorContext.getTermInformation().update(1, "leader");
1669
1670         leader = new Leader(leaderActorContext);
1671
1672         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1673
1674         assertEquals(payloadVersion, leader.getLeaderPayloadVersion());
1675         assertEquals(RaftVersions.HELIUM_VERSION, followerInfo.getRaftVersion());
1676
1677         AppendEntriesReply reply = new AppendEntriesReply(FOLLOWER_ID, 1, true, 2, 1, payloadVersion);
1678
1679         RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
1680
1681         assertEquals(RaftState.Leader, raftActorBehavior.state());
1682
1683         assertEquals(2, leaderActorContext.getCommitIndex());
1684
1685         ApplyJournalEntries applyJournalEntries = MessageCollectorActor.expectFirstMatching(
1686                 leaderActor, ApplyJournalEntries.class);
1687
1688         assertEquals(2, leaderActorContext.getLastApplied());
1689
1690         assertEquals(2, applyJournalEntries.getToIndex());
1691
1692         List<ApplyState> applyStateList = MessageCollectorActor.getAllMatching(leaderActor,
1693                 ApplyState.class);
1694
1695         assertEquals(1,applyStateList.size());
1696
1697         ApplyState applyState = applyStateList.get(0);
1698
1699         assertEquals(2, applyState.getReplicatedLogEntry().getIndex());
1700
1701         assertEquals(2, followerInfo.getMatchIndex());
1702         assertEquals(3, followerInfo.getNextIndex());
1703         assertEquals(payloadVersion, followerInfo.getPayloadVersion());
1704         assertEquals(RaftVersions.CURRENT_VERSION, followerInfo.getRaftVersion());
1705     }
1706
1707     @Test
1708     public void testHandleAppendEntriesReplyUnknownFollower() {
1709         logStart("testHandleAppendEntriesReplyUnknownFollower");
1710
1711         MockRaftActorContext leaderActorContext = createActorContext();
1712
1713         leader = new Leader(leaderActorContext);
1714
1715         AppendEntriesReply reply = new AppendEntriesReply("unkown-follower", 1, false, 10, 1, (short)0);
1716
1717         RaftActorBehavior raftActorBehavior = leader.handleAppendEntriesReply(followerActor, reply);
1718
1719         assertEquals(RaftState.Leader, raftActorBehavior.state());
1720     }
1721
1722     @Test
1723     public void testFollowerCatchUpWithAppendEntriesMaxDataSizeExceeded() {
1724         logStart("testFollowerCatchUpWithAppendEntriesMaxDataSizeExceeded");
1725
1726         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1727         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1728                 new FiniteDuration(1000, TimeUnit.SECONDS));
1729         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setSnapshotChunkSize(2);
1730
1731         leaderActorContext.setReplicatedLog(
1732                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 4, 1).build());
1733         long leaderCommitIndex = 3;
1734         leaderActorContext.setCommitIndex(leaderCommitIndex);
1735         leaderActorContext.setLastApplied(leaderCommitIndex);
1736
1737         final ReplicatedLogEntry leadersFirstLogEntry = leaderActorContext.getReplicatedLog().get(0);
1738         final ReplicatedLogEntry leadersSecondLogEntry = leaderActorContext.getReplicatedLog().get(1);
1739         final ReplicatedLogEntry leadersThirdLogEntry = leaderActorContext.getReplicatedLog().get(2);
1740         final ReplicatedLogEntry leadersFourthLogEntry = leaderActorContext.getReplicatedLog().get(3);
1741
1742         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1743
1744         followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
1745         followerActorContext.setCommitIndex(-1);
1746         followerActorContext.setLastApplied(-1);
1747
1748         Follower follower = new Follower(followerActorContext);
1749         followerActor.underlyingActor().setBehavior(follower);
1750         followerActorContext.setCurrentBehavior(follower);
1751
1752         leader = new Leader(leaderActorContext);
1753
1754         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
1755         final AppendEntriesReply appendEntriesReply = MessageCollectorActor.expectFirstMatching(leaderActor,
1756                 AppendEntriesReply.class);
1757
1758         MessageCollectorActor.clearMessages(followerActor);
1759         MessageCollectorActor.clearMessages(leaderActor);
1760
1761         // Verify initial AppendEntries sent with the leader's current commit index.
1762         assertEquals("getLeaderCommit", -1, appendEntries.getLeaderCommit());
1763         assertEquals("Log entries size", 0, appendEntries.getEntries().size());
1764         assertEquals("getPrevLogIndex", 2, appendEntries.getPrevLogIndex());
1765
1766         leaderActor.underlyingActor().setBehavior(leader);
1767         leaderActorContext.setCurrentBehavior(leader);
1768
1769         leader.handleMessage(followerActor, appendEntriesReply);
1770
1771         List<AppendEntries> appendEntriesList = MessageCollectorActor.expectMatching(followerActor,
1772                 AppendEntries.class, 2);
1773         MessageCollectorActor.expectMatching(leaderActor, AppendEntriesReply.class, 2);
1774
1775         appendEntries = appendEntriesList.get(0);
1776         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1777         assertEquals("getPrevLogIndex", -1, appendEntries.getPrevLogIndex());
1778         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1779
1780         assertEquals("First entry index", 0, appendEntries.getEntries().get(0).getIndex());
1781         assertEquals("First entry data", leadersFirstLogEntry.getData(),
1782                 appendEntries.getEntries().get(0).getData());
1783         assertEquals("Second entry index", 1, appendEntries.getEntries().get(1).getIndex());
1784         assertEquals("Second entry data", leadersSecondLogEntry.getData(),
1785                 appendEntries.getEntries().get(1).getData());
1786
1787         appendEntries = appendEntriesList.get(1);
1788         assertEquals("getLeaderCommit", leaderCommitIndex, appendEntries.getLeaderCommit());
1789         assertEquals("getPrevLogIndex", 1, appendEntries.getPrevLogIndex());
1790         assertEquals("Log entries size", 2, appendEntries.getEntries().size());
1791
1792         assertEquals("First entry index", 2, appendEntries.getEntries().get(0).getIndex());
1793         assertEquals("First entry data", leadersThirdLogEntry.getData(),
1794                 appendEntries.getEntries().get(0).getData());
1795         assertEquals("Second entry index", 3, appendEntries.getEntries().get(1).getIndex());
1796         assertEquals("Second entry data", leadersFourthLogEntry.getData(),
1797                 appendEntries.getEntries().get(1).getData());
1798
1799         FollowerLogInformation followerInfo = leader.getFollower(FOLLOWER_ID);
1800         assertEquals("getNextIndex", 4, followerInfo.getNextIndex());
1801
1802         MessageCollectorActor.expectMatching(followerActor, ApplyState.class, 4);
1803
1804         assertEquals("Follower's commit index", 3, followerActorContext.getCommitIndex());
1805         assertEquals("Follower's lastIndex", 3, followerActorContext.getReplicatedLog().lastIndex());
1806     }
1807
1808     @Test
1809     public void testHandleRequestVoteReply() {
1810         logStart("testHandleRequestVoteReply");
1811
1812         MockRaftActorContext leaderActorContext = createActorContext();
1813
1814         leader = new Leader(leaderActorContext);
1815
1816         // Should be a no-op.
1817         RaftActorBehavior raftActorBehavior = leader.handleRequestVoteReply(followerActor,
1818                 new RequestVoteReply(1, true));
1819
1820         assertEquals(RaftState.Leader, raftActorBehavior.state());
1821
1822         raftActorBehavior = leader.handleRequestVoteReply(followerActor, new RequestVoteReply(1, false));
1823
1824         assertEquals(RaftState.Leader, raftActorBehavior.state());
1825     }
1826
1827     @Test
1828     public void testIsolatedLeaderCheckNoFollowers() {
1829         logStart("testIsolatedLeaderCheckNoFollowers");
1830
1831         MockRaftActorContext leaderActorContext = createActorContext();
1832
1833         leader = new Leader(leaderActorContext);
1834         RaftActorBehavior newBehavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1835         assertTrue(newBehavior instanceof Leader);
1836     }
1837
1838     @Test
1839     public void testIsolatedLeaderCheckNoVotingFollowers() {
1840         logStart("testIsolatedLeaderCheckNoVotingFollowers");
1841
1842         MockRaftActorContext followerActorContext = createFollowerActorContextWithLeader();
1843         Follower follower = new Follower(followerActorContext);
1844         followerActor.underlyingActor().setBehavior(follower);
1845
1846         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1847         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1848                 new FiniteDuration(1000, TimeUnit.SECONDS));
1849         leaderActorContext.getPeerInfo(FOLLOWER_ID).setVotingState(VotingState.NON_VOTING);
1850
1851         leader = new Leader(leaderActorContext);
1852         leader.getFollower(FOLLOWER_ID).markFollowerActive();
1853         RaftActorBehavior newBehavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1854         assertTrue("Expected Leader", newBehavior instanceof Leader);
1855     }
1856
1857     private RaftActorBehavior setupIsolatedLeaderCheckTestWithTwoFollowers(RaftPolicy raftPolicy) {
1858         ActorRef followerActor1 = getSystem().actorOf(MessageCollectorActor.props(), "follower-1");
1859         ActorRef followerActor2 = getSystem().actorOf(MessageCollectorActor.props(), "follower-2");
1860
1861         MockRaftActorContext leaderActorContext = createActorContext();
1862
1863         Map<String, String> peerAddresses = new HashMap<>();
1864         peerAddresses.put("follower-1", followerActor1.path().toString());
1865         peerAddresses.put("follower-2", followerActor2.path().toString());
1866
1867         leaderActorContext.setPeerAddresses(peerAddresses);
1868         leaderActorContext.setRaftPolicy(raftPolicy);
1869
1870         leader = new Leader(leaderActorContext);
1871
1872         leader.markFollowerActive("follower-1");
1873         leader.markFollowerActive("follower-2");
1874         RaftActorBehavior newBehavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1875         assertTrue("Behavior not instance of Leader when all followers are active", newBehavior instanceof Leader);
1876
1877         // kill 1 follower and verify if that got killed
1878         final JavaTestKit probe = new JavaTestKit(getSystem());
1879         probe.watch(followerActor1);
1880         followerActor1.tell(PoisonPill.getInstance(), ActorRef.noSender());
1881         final Terminated termMsg1 = probe.expectMsgClass(Terminated.class);
1882         assertEquals(termMsg1.getActor(), followerActor1);
1883
1884         leader.markFollowerInActive("follower-1");
1885         leader.markFollowerActive("follower-2");
1886         newBehavior = leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1887         assertTrue("Behavior not instance of Leader when majority of followers are active",
1888                 newBehavior instanceof Leader);
1889
1890         // kill 2nd follower and leader should change to Isolated leader
1891         followerActor2.tell(PoisonPill.getInstance(), null);
1892         probe.watch(followerActor2);
1893         followerActor2.tell(PoisonPill.getInstance(), ActorRef.noSender());
1894         final Terminated termMsg2 = probe.expectMsgClass(Terminated.class);
1895         assertEquals(termMsg2.getActor(), followerActor2);
1896
1897         leader.markFollowerInActive("follower-2");
1898         return leader.handleMessage(leaderActor, Leader.ISOLATED_LEADER_CHECK);
1899     }
1900
1901     @Test
1902     public void testIsolatedLeaderCheckTwoFollowers() throws Exception {
1903         logStart("testIsolatedLeaderCheckTwoFollowers");
1904
1905         RaftActorBehavior newBehavior = setupIsolatedLeaderCheckTestWithTwoFollowers(DefaultRaftPolicy.INSTANCE);
1906
1907         assertTrue("Behavior not instance of IsolatedLeader when majority followers are inactive",
1908             newBehavior instanceof IsolatedLeader);
1909     }
1910
1911     @Test
1912     public void testIsolatedLeaderCheckTwoFollowersWhenElectionsAreDisabled() throws Exception {
1913         logStart("testIsolatedLeaderCheckTwoFollowersWhenElectionsAreDisabled");
1914
1915         RaftActorBehavior newBehavior = setupIsolatedLeaderCheckTestWithTwoFollowers(createRaftPolicy(false, true));
1916
1917         assertTrue("Behavior should not switch to IsolatedLeader because elections are disabled",
1918                 newBehavior instanceof Leader);
1919     }
1920
1921     @Test
1922     public void testLaggingFollowerStarvation() throws Exception {
1923         logStart("testLaggingFollowerStarvation");
1924
1925         String leaderActorId = actorFactory.generateActorId("leader");
1926         String follower1ActorId = actorFactory.generateActorId("follower");
1927         String follower2ActorId = actorFactory.generateActorId("follower");
1928
1929         final ActorRef follower1Actor = actorFactory.createActor(MessageCollectorActor.props(), follower1ActorId);
1930         final ActorRef follower2Actor = actorFactory.createActor(MessageCollectorActor.props(), follower2ActorId);
1931
1932         MockRaftActorContext leaderActorContext =
1933                 new MockRaftActorContext(leaderActorId, getSystem(), leaderActor);
1934
1935         DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl();
1936         configParams.setHeartBeatInterval(new FiniteDuration(200, TimeUnit.MILLISECONDS));
1937         configParams.setIsolatedLeaderCheckInterval(new FiniteDuration(10, TimeUnit.SECONDS));
1938
1939         leaderActorContext.setConfigParams(configParams);
1940
1941         leaderActorContext.setReplicatedLog(
1942                 new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(1,5,1).build());
1943
1944         Map<String, String> peerAddresses = new HashMap<>();
1945         peerAddresses.put(follower1ActorId,
1946                 follower1Actor.path().toString());
1947         peerAddresses.put(follower2ActorId,
1948                 follower2Actor.path().toString());
1949
1950         leaderActorContext.setPeerAddresses(peerAddresses);
1951         leaderActorContext.getTermInformation().update(1, leaderActorId);
1952
1953         leader = createBehavior(leaderActorContext);
1954
1955         leaderActor.underlyingActor().setBehavior(leader);
1956
1957         for (int i = 1; i < 6; i++) {
1958             // Each AppendEntriesReply could end up rescheduling the heartbeat (without the fix for bug 2733)
1959             RaftActorBehavior newBehavior = leader.handleMessage(follower1Actor,
1960                     new AppendEntriesReply(follower1ActorId, 1, true, i, 1, (short)0));
1961             assertTrue(newBehavior == leader);
1962             Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
1963         }
1964
1965         // Check if the leader has been receiving SendHeartbeat messages despite getting AppendEntriesReply
1966         List<SendHeartBeat> heartbeats = MessageCollectorActor.getAllMatching(leaderActor, SendHeartBeat.class);
1967
1968         assertTrue(String.format("%s heartbeat(s) is less than expected", heartbeats.size()),
1969                 heartbeats.size() > 1);
1970
1971         // Check if follower-2 got AppendEntries during this time and was not starved
1972         List<AppendEntries> appendEntries = MessageCollectorActor.getAllMatching(follower2Actor, AppendEntries.class);
1973
1974         assertTrue(String.format("%s append entries is less than expected", appendEntries.size()),
1975                 appendEntries.size() > 1);
1976     }
1977
1978     @Test
1979     public void testReplicationConsensusWithNonVotingFollower() {
1980         logStart("testReplicationConsensusWithNonVotingFollower");
1981
1982         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
1983         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
1984                 new FiniteDuration(1000, TimeUnit.SECONDS));
1985
1986         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
1987         leaderActorContext.setCommitIndex(-1);
1988         leaderActorContext.setLastApplied(-1);
1989
1990         String nonVotingFollowerId = "nonvoting-follower";
1991         TestActorRef<ForwardMessageToBehaviorActor> nonVotingFollowerActor = actorFactory.createTestActor(
1992                 Props.create(MessageCollectorActor.class), actorFactory.generateActorId(nonVotingFollowerId));
1993
1994         leaderActorContext.addToPeers(nonVotingFollowerId, nonVotingFollowerActor.path().toString(),
1995                 VotingState.NON_VOTING);
1996
1997         leader = new Leader(leaderActorContext);
1998         leaderActorContext.setCurrentBehavior(leader);
1999
2000         // Ignore initial heartbeats
2001         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2002         MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor, AppendEntries.class);
2003
2004         MessageCollectorActor.clearMessages(followerActor);
2005         MessageCollectorActor.clearMessages(nonVotingFollowerActor);
2006         MessageCollectorActor.clearMessages(leaderActor);
2007
2008         // Send a Replicate message and wait for AppendEntries.
2009         sendReplicate(leaderActorContext, 0);
2010
2011         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2012         MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor, AppendEntries.class);
2013
2014         // Send reply only from the voting follower and verify consensus via ApplyState.
2015         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 0, 1, (short)0));
2016
2017         MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class);
2018
2019         leader.handleMessage(leaderActor, new AppendEntriesReply(nonVotingFollowerId, 1, true, 0, 1, (short)0));
2020
2021         MessageCollectorActor.clearMessages(followerActor);
2022         MessageCollectorActor.clearMessages(nonVotingFollowerActor);
2023         MessageCollectorActor.clearMessages(leaderActor);
2024
2025         // Send another Replicate message
2026         sendReplicate(leaderActorContext, 1);
2027
2028         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2029         AppendEntries appendEntries = MessageCollectorActor.expectFirstMatching(nonVotingFollowerActor,
2030                 AppendEntries.class);
2031         assertEquals("Log entries size", 1, appendEntries.getEntries().size());
2032         assertEquals("Log entry index", 1, appendEntries.getEntries().get(0).getIndex());
2033
2034         // Send reply only from the non-voting follower and verify no consensus via no ApplyState.
2035         leader.handleMessage(leaderActor, new AppendEntriesReply(nonVotingFollowerId, 1, true, 1, 1, (short)0));
2036
2037         MessageCollectorActor.assertNoneMatching(leaderActor, ApplyState.class, 500);
2038
2039         // Send reply from the voting follower and verify consensus.
2040         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 1, 1, (short)0));
2041
2042         MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class);
2043     }
2044
2045     @Test
2046     public void testTransferLeadershipWithFollowerInSync() {
2047         logStart("testTransferLeadershipWithFollowerInSync");
2048
2049         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2050         leaderActorContext.setLastApplied(-1);
2051         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2052                 new FiniteDuration(1000, TimeUnit.SECONDS));
2053         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
2054
2055         leader = new Leader(leaderActorContext);
2056         leaderActorContext.setCurrentBehavior(leader);
2057
2058         // Initial heartbeat
2059         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2060         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2061         MessageCollectorActor.clearMessages(followerActor);
2062
2063         sendReplicate(leaderActorContext, 0);
2064         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2065
2066         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 0, 1, (short)0));
2067         MessageCollectorActor.expectFirstMatching(leaderActor, ApplyState.class);
2068         MessageCollectorActor.clearMessages(followerActor);
2069
2070         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2071         leader.transferLeadership(mockTransferCohort);
2072
2073         verify(mockTransferCohort, never()).transferComplete();
2074         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2075         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 0, 1, (short)0));
2076
2077         // Expect a final AppendEntries to ensure the follower's lastApplied index is up-to-date
2078         MessageCollectorActor.expectMatching(followerActor, AppendEntries.class, 2);
2079
2080         // Leader should force an election timeout
2081         MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
2082
2083         verify(mockTransferCohort).transferComplete();
2084     }
2085
2086     @Test
2087     public void testTransferLeadershipWithEmptyLog() {
2088         logStart("testTransferLeadershipWithEmptyLog");
2089
2090         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2091         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2092                 new FiniteDuration(1000, TimeUnit.SECONDS));
2093         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
2094
2095         leader = new Leader(leaderActorContext);
2096         leaderActorContext.setCurrentBehavior(leader);
2097
2098         // Initial heartbeat
2099         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2100         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2101         MessageCollectorActor.clearMessages(followerActor);
2102
2103         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2104         leader.transferLeadership(mockTransferCohort);
2105
2106         verify(mockTransferCohort, never()).transferComplete();
2107         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2108         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2109
2110         // Expect a final AppendEntries to ensure the follower's lastApplied index is up-to-date
2111         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2112
2113         // Leader should force an election timeout
2114         MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
2115
2116         verify(mockTransferCohort).transferComplete();
2117     }
2118
2119     @Test
2120     public void testTransferLeadershipWithFollowerInitiallyOutOfSync() {
2121         logStart("testTransferLeadershipWithFollowerInitiallyOutOfSync");
2122
2123         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2124         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2125                 new FiniteDuration(200, TimeUnit.MILLISECONDS));
2126
2127         leader = new Leader(leaderActorContext);
2128         leaderActorContext.setCurrentBehavior(leader);
2129
2130         // Initial heartbeat
2131         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2132         MessageCollectorActor.clearMessages(followerActor);
2133
2134         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2135         leader.transferLeadership(mockTransferCohort);
2136
2137         verify(mockTransferCohort, never()).transferComplete();
2138
2139         // Sync up the follower.
2140         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2141         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2142         MessageCollectorActor.clearMessages(followerActor);
2143
2144         Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams()
2145                 .getHeartBeatInterval().toMillis() + 1, TimeUnit.MILLISECONDS);
2146         leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
2147         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2148         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, 1, 1, (short)0));
2149
2150         // Leader should force an election timeout
2151         MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
2152
2153         verify(mockTransferCohort).transferComplete();
2154     }
2155
2156     @Test
2157     public void testTransferLeadershipWithFollowerSyncTimeout() {
2158         logStart("testTransferLeadershipWithFollowerSyncTimeout");
2159
2160         MockRaftActorContext leaderActorContext = createActorContextWithFollower();
2161         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setHeartBeatInterval(
2162                 new FiniteDuration(200, TimeUnit.MILLISECONDS));
2163         ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(2);
2164         leaderActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build());
2165
2166         leader = new Leader(leaderActorContext);
2167         leaderActorContext.setCurrentBehavior(leader);
2168
2169         // Initial heartbeat
2170         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2171         leader.handleMessage(leaderActor, new AppendEntriesReply(FOLLOWER_ID, 1, true, -1, -1, (short)0));
2172         MessageCollectorActor.clearMessages(followerActor);
2173
2174         sendReplicate(leaderActorContext, 0);
2175         MessageCollectorActor.expectFirstMatching(followerActor, AppendEntries.class);
2176
2177         MessageCollectorActor.clearMessages(followerActor);
2178
2179         RaftActorLeadershipTransferCohort mockTransferCohort = mock(RaftActorLeadershipTransferCohort.class);
2180         leader.transferLeadership(mockTransferCohort);
2181
2182         verify(mockTransferCohort, never()).transferComplete();
2183
2184         // Send heartbeats to time out the transfer.
2185         for (int i = 0; i < leaderActorContext.getConfigParams().getElectionTimeoutFactor(); i++) {
2186             Uninterruptibles.sleepUninterruptibly(leaderActorContext.getConfigParams()
2187                     .getHeartBeatInterval().toMillis() + 1, TimeUnit.MILLISECONDS);
2188             leader.handleMessage(leaderActor, SendHeartBeat.INSTANCE);
2189         }
2190
2191         verify(mockTransferCohort).abortTransfer();
2192         verify(mockTransferCohort, never()).transferComplete();
2193         MessageCollectorActor.assertNoneMatching(followerActor, ElectionTimeout.class, 100);
2194     }
2195
2196     @Override
2197     protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(MockRaftActorContext actorContext,
2198             ActorRef actorRef, RaftRPC rpc) throws Exception {
2199         super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
2200         assertEquals("New votedFor", null, actorContext.getTermInformation().getVotedFor());
2201     }
2202
2203     private class MockConfigParamsImpl extends DefaultConfigParamsImpl {
2204
2205         private final long electionTimeOutIntervalMillis;
2206         private final int snapshotChunkSize;
2207
2208         MockConfigParamsImpl(long electionTimeOutIntervalMillis, int snapshotChunkSize) {
2209             super();
2210             this.electionTimeOutIntervalMillis = electionTimeOutIntervalMillis;
2211             this.snapshotChunkSize = snapshotChunkSize;
2212         }
2213
2214         @Override
2215         public FiniteDuration getElectionTimeOutInterval() {
2216             return new FiniteDuration(electionTimeOutIntervalMillis, TimeUnit.MILLISECONDS);
2217         }
2218
2219         @Override
2220         public int getSnapshotChunkSize() {
2221             return snapshotChunkSize;
2222         }
2223     }
2224 }