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

   1 /*
   2  * Copyright (c) 2015 Brocade Communications 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 package org.opendaylight.controller.cluster.raft;
   9
  10 import static org.junit.Assert.assertEquals;
  11 import static org.junit.Assert.assertNotNull;
  12
  13 import akka.actor.ActorRef;
  14 import akka.persistence.SaveSnapshotSuccess;
  15 import com.google.common.collect.ImmutableMap;
  16 import com.google.common.util.concurrent.Uninterruptibles;
  17 import java.util.Arrays;
  18 import java.util.HashSet;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.Set;
  22 import java.util.concurrent.TimeUnit;
  23 import org.apache.commons.lang3.SerializationUtils;
  24 import org.eclipse.jdt.annotation.Nullable;
  25 import org.junit.Assert;
  26 import org.junit.Test;
  27 import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload;
  28 import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
  29 import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
  30 import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
  31 import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader;
  32 import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
  33 import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
  34 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
  35 import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
  36 import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
  37 import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries;
  38 import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
  39 import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
  40 import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
  41 import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
  42 import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
  43 import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal;
  44 import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore;
  45 import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
  46
  47 /**
  48  * Tests replication and snapshots end-to-end using real RaftActors and behavior communication with a
  49  * lagging follower.
  50  *
  51  * @author Thomas Pantelis
  52  */
  53 public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends AbstractRaftActorIntegrationTest {
  54
  55     private void setup() {
  56         leaderId = factory.generateActorId("leader");
  57         follower1Id = factory.generateActorId("follower");
  58         follower2Id = factory.generateActorId("follower");
  59
  60         // Setup the persistent journal for the leader - just an election term and no journal/snapshots.
  61         InMemoryJournal.addEntry(leaderId, 1, new UpdateElectionTerm(initialTerm, leaderId));
  62
  63         // Create the leader and 2 follower actors.
  64         follower1Actor = newTestRaftActor(follower1Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
  65                 follower2Id, testActorPath(follower2Id)), newFollowerConfigParams());
  66
  67         follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
  68                 follower1Id, testActorPath(follower1Id)), newFollowerConfigParams());
  69
  70         Map<String, String> leaderPeerAddresses = ImmutableMap.<String, String>builder()
  71                 .put(follower1Id, follower1Actor.path().toString())
  72                 .put(follower2Id, follower2Actor.path().toString()).build();
  73
  74         leaderConfigParams = newLeaderConfigParams();
  75         leaderActor = newTestRaftActor(leaderId, leaderPeerAddresses, leaderConfigParams);
  76
  77         waitUntilLeader(leaderActor);
  78
  79         leaderContext = leaderActor.underlyingActor().getRaftActorContext();
  80         leader = leaderActor.underlyingActor().getCurrentBehavior();
  81
  82         follower1Context = follower1Actor.underlyingActor().getRaftActorContext();
  83         follower1 = follower1Actor.underlyingActor().getCurrentBehavior();
  84
  85         follower2Context = follower2Actor.underlyingActor().getRaftActorContext();
  86         follower2 = follower2Actor.underlyingActor().getCurrentBehavior();
  87
  88         currentTerm = leaderContext.getTermInformation().getCurrentTerm();
  89         assertEquals("Current term > " + initialTerm, true, currentTerm > initialTerm);
  90
  91         leaderCollectorActor = leaderActor.underlyingActor().collectorActor();
  92         follower1CollectorActor = follower1Actor.underlyingActor().collectorActor();
  93         follower2CollectorActor = follower2Actor.underlyingActor().collectorActor();
  94
  95         testLog.info("Leader created and elected");
  96     }
  97
  98     private void setupFollower2() {
  99         follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId),
 100                 follower1Id, testActorPath(follower1Id)), newFollowerConfigParams());
 101
 102         follower2Context = follower2Actor.underlyingActor().getRaftActorContext();
 103         follower2 = follower2Actor.underlyingActor().getCurrentBehavior();
 104
 105         follower2CollectorActor = follower2Actor.underlyingActor().collectorActor();
 106     }
 107
 108     /**
 109      * Send 2 payload instances with follower 2 lagging then resume the follower and verifies it gets
 110      * caught up via AppendEntries.
 111      */
 112     @Test
 113     public void testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries() {
 114         testLog.info("testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries starting: sending 2 new payloads");
 115
 116         setup();
 117
 118         // Simulate lagging by dropping AppendEntries messages in follower 2.
 119         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 120
 121         // Send the payloads.
 122         MockPayload payload0 = sendPayloadData(leaderActor, "zero");
 123         MockPayload payload1 = sendPayloadData(leaderActor, "one");
 124
 125         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 126         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 2);
 127         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload0.toString(), currentTerm, 0, payload0);
 128         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload1.toString(), currentTerm, 1, payload1);
 129
 130         // Verify follower 1 applies each log entry.
 131         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 2);
 132         verifyApplyState(applyStates.get(0), null, null, currentTerm, 0, payload0);
 133         verifyApplyState(applyStates.get(1), null, null, currentTerm, 1, payload1);
 134
 135         // Ensure there's at least 1 more heartbeat.
 136         MessageCollectorActor.clearMessages(leaderCollectorActor);
 137         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 138
 139         // The leader should not have performed fake snapshots to trim the log because the entries have not
 140         // been replicated to follower 2.
 141         assertEquals("Leader snapshot term", -1, leaderContext.getReplicatedLog().getSnapshotTerm());
 142         assertEquals("Leader snapshot index", -1, leaderContext.getReplicatedLog().getSnapshotIndex());
 143         assertEquals("Leader journal log size", 2, leaderContext.getReplicatedLog().size());
 144         assertEquals("Leader journal last index", 1, leaderContext.getReplicatedLog().lastIndex());
 145         assertEquals("Leader commit index", 1, leaderContext.getCommitIndex());
 146         assertEquals("Leader last applied", 1, leaderContext.getLastApplied());
 147         assertEquals("Leader replicatedToAllIndex", -1, leader.getReplicatedToAllIndex());
 148
 149         testLog.info(
 150             "testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries: new entries applied - resuming follower {}",
 151             follower2Id);
 152
 153         // Now stop dropping AppendEntries in follower 2.
 154         follower2Actor.underlyingActor().stopDropMessages(AppendEntries.class);
 155
 156         // Verify follower 2 applies each log entry.
 157         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 2);
 158         verifyApplyState(applyStates.get(0), null, null, currentTerm, 0, payload0);
 159         verifyApplyState(applyStates.get(1), null, null, currentTerm, 1, payload1);
 160
 161         // Ensure there's at least 1 more heartbeat.
 162         MessageCollectorActor.clearMessages(leaderCollectorActor);
 163         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 164
 165         // The leader should now have performed fake snapshots to trim the log.
 166         verifyLeadersTrimmedLog(1);
 167
 168         // Even though follower 2 lagged behind, the leader should not have tried to install a snapshot
 169         // to catch it up because no snapshotting was done so the follower's next index was present in the log.
 170         InstallSnapshot installSnapshot = MessageCollectorActor.getFirstMatching(follower2CollectorActor,
 171                 InstallSnapshot.class);
 172         Assert.assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot);
 173
 174         testLog.info("testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries complete");
 175     }
 176
 177     /**
 178      * Send payloads to trigger a leader snapshot due to snapshotBatchCount reached with follower 2
 179      * lagging but not enough for the leader to trim its log from the last applied index. Follower 2's log
 180      * will be behind by several entries and, when it is resumed, it should be caught up via AppendEntries
 181      * sent by the leader.
 182      */
 183     @Test
 184     public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries() {
 185         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries starting");
 186
 187         setup();
 188
 189         sendInitialPayloadsReplicatedToAllFollowers("zero", "one");
 190
 191         // Configure follower 2 to drop messages and lag.
 192         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 193
 194         // Send the first payload and verify it gets applied by the leader and follower 1.
 195         MockPayload payload2 = sendPayloadData(leaderActor, "two");
 196
 197         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 198         verifyApplyState(applyState, leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2);
 199
 200         applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class);
 201         verifyApplyState(applyState, null, null, currentTerm, 2, payload2);
 202
 203         expSnapshotState.add(payload2);
 204
 205         MessageCollectorActor.clearMessages(leaderCollectorActor);
 206         MessageCollectorActor.clearMessages(follower1CollectorActor);
 207
 208         // Send another payload - this should cause a snapshot due to snapshotBatchCount reached.
 209         MockPayload payload3 = sendPayloadData(leaderActor, "three");
 210
 211         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 212
 213         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries: sending 2 more payloads");
 214
 215         // Send 2 more payloads - not enough to trigger another snapshot.
 216         MockPayload payload4 = sendPayloadData(leaderActor, "four");
 217         MockPayload payload5 = sendPayloadData(leaderActor, "five");
 218
 219         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 220         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 3);
 221         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload3.toString(), currentTerm, 3, payload3);
 222         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
 223         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload5.toString(), currentTerm, 5, payload5);
 224
 225         // Verify follower 1 applies each log entry.
 226         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 3);
 227         verifyApplyState(applyStates.get(0), null, null, currentTerm, 3, payload3);
 228         verifyApplyState(applyStates.get(1), null, null, currentTerm, 4, payload4);
 229         verifyApplyState(applyStates.get(2), null, null, currentTerm, 5, payload5);
 230
 231         // The snapshot should have caused the leader to advanced the snapshot index to the
 232         // last previously applied index (1) that was replicated to all followers at the time of capture.
 233         // Note: since the log size (3) did not exceed the snapshot batch count (4), the leader should not
 234         // have trimmed the log to the last index actually applied (5).
 235         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
 236         assertEquals("Leader snapshot index", 1, leaderContext.getReplicatedLog().getSnapshotIndex());
 237         assertEquals("Leader journal log size", 4, leaderContext.getReplicatedLog().size());
 238         assertEquals("Leader journal last index", 5, leaderContext.getReplicatedLog().lastIndex());
 239         assertEquals("Leader commit index", 5, leaderContext.getCommitIndex());
 240         assertEquals("Leader last applied", 5, leaderContext.getLastApplied());
 241         assertEquals("Leader replicatedToAllIndex", 1, leader.getReplicatedToAllIndex());
 242
 243         // Now stop dropping AppendEntries in follower 2.
 244         follower2Actor.underlyingActor().stopDropMessages(AppendEntries.class);
 245
 246         // Verify follower 2 applies each log entry. The leader should not install a snapshot b/c
 247         // follower 2's next index (3) is still present in the log.
 248         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 4);
 249         verifyApplyState(applyStates.get(0), null, null, currentTerm, 2, payload2);
 250         verifyApplyState(applyStates.get(1), null, null, currentTerm, 3, payload3);
 251         verifyApplyState(applyStates.get(2), null, null, currentTerm, 4, payload4);
 252         verifyApplyState(applyStates.get(3), null, null, currentTerm, 5, payload5);
 253
 254         // Verify the leader did not try to install a snapshot to catch up follower 2.
 255         InstallSnapshot installSnapshot = MessageCollectorActor.getFirstMatching(follower2CollectorActor,
 256                 InstallSnapshot.class);
 257         Assert.assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot);
 258
 259         // Ensure there's at least 1 more heartbeat.
 260         MessageCollectorActor.clearMessages(leaderCollectorActor);
 261         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 262
 263         // The leader should now have performed fake snapshots to advance the snapshot index and to trim
 264         // the log. In addition replicatedToAllIndex should've advanced.
 265         verifyLeadersTrimmedLog(5);
 266
 267         // Verify the leader's persisted snapshot.
 268         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 269         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
 270         verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 2, currentTerm, 3);
 271         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
 272         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 273         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 3, payload3);
 274
 275         // Verify follower 1's log and snapshot indexes.
 276         MessageCollectorActor.clearMessages(follower1CollectorActor);
 277         MessageCollectorActor.expectFirstMatching(follower1CollectorActor, AppendEntries.class);
 278         verifyFollowersTrimmedLog(1, follower1Actor, 5);
 279
 280         // Verify follower 2's log and snapshot indexes.
 281         MessageCollectorActor.clearMessages(follower2CollectorActor);
 282         MessageCollectorActor.expectFirstMatching(follower2CollectorActor, AppendEntries.class);
 283         verifyFollowersTrimmedLog(2, follower2Actor, 5);
 284
 285         MessageCollectorActor.clearMessages(leaderCollectorActor);
 286         MessageCollectorActor.clearMessages(follower1CollectorActor);
 287         MessageCollectorActor.clearMessages(follower2CollectorActor);
 288
 289         expSnapshotState.add(payload3);
 290         expSnapshotState.add(payload4);
 291         expSnapshotState.add(payload5);
 292
 293         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries complete");
 294     }
 295
 296     /**
 297      * Send payloads to trigger a leader snapshot due to snapshotBatchCount reached with follower 2
 298      * lagging where the leader trims its log from the last applied index. Follower 2's log
 299      * will be behind by several entries and, when it is resumed, it should be caught up via a snapshot
 300      * installed by the leader.
 301      */
 302     @Test
 303     public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot() {
 304         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot starting");
 305
 306         setup();
 307
 308         sendInitialPayloadsReplicatedToAllFollowers("zero", "one");
 309
 310         // Configure follower 2 to drop messages and lag.
 311         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 312
 313         // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader.
 314         Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5,
 315                 TimeUnit.MILLISECONDS);
 316
 317         // Send 5 payloads - the second should cause a leader snapshot.
 318         final MockPayload payload2 = sendPayloadData(leaderActor, "two");
 319         final MockPayload payload3 = sendPayloadData(leaderActor, "three");
 320         final MockPayload payload4 = sendPayloadData(leaderActor, "four");
 321         final MockPayload payload5 = sendPayloadData(leaderActor, "five");
 322         final MockPayload payload6 = sendPayloadData(leaderActor, "six");
 323
 324         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 325
 326         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 327         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 5);
 328         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2);
 329         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
 330         verifyApplyState(applyStates.get(4), leaderCollectorActor, payload6.toString(), currentTerm, 6, payload6);
 331
 332         MessageCollectorActor.clearMessages(leaderCollectorActor);
 333
 334         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot: "
 335                 + "sending 1 more payload to trigger second snapshot");
 336
 337         // Send another payload to trigger a second leader snapshot.
 338         MockPayload payload7 = sendPayloadData(leaderActor, "seven");
 339
 340         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 341
 342         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 343         verifyApplyState(applyState, leaderCollectorActor, payload7.toString(), currentTerm, 7, payload7);
 344
 345         // Verify follower 1 applies each log entry.
 346         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 6);
 347         verifyApplyState(applyStates.get(0), null, null, currentTerm, 2, payload2);
 348         verifyApplyState(applyStates.get(2), null, null, currentTerm, 4, payload4);
 349         verifyApplyState(applyStates.get(5), null, null, currentTerm, 7, payload7);
 350
 351         // The snapshot should have caused the leader to advanced the snapshot index to the leader's last
 352         // applied index (6) since the log size should have exceed the snapshot batch count (4).
 353         // replicatedToAllIndex should remain at 1 since follower 2 is lagging.
 354         verifyLeadersTrimmedLog(7, 1);
 355
 356         expSnapshotState.add(payload2);
 357         expSnapshotState.add(payload3);
 358         expSnapshotState.add(payload4);
 359         expSnapshotState.add(payload5);
 360         expSnapshotState.add(payload6);
 361
 362         MessageCollectorActor.clearMessages(leaderCollectorActor);
 363         MessageCollectorActor.clearMessages(follower1CollectorActor);
 364
 365         // Send a server config change to test that the install snapshot includes the server config.
 366
 367         ServerConfigurationPayload serverConfig = new ServerConfigurationPayload(Arrays.asList(
 368                 new ServerInfo(leaderId, true),
 369                 new ServerInfo(follower1Id, false),
 370                 new ServerInfo(follower2Id, false)));
 371         leaderContext.updatePeerIds(serverConfig);
 372         ((AbstractLeader)leader).updateMinReplicaCount();
 373         leaderActor.tell(serverConfig, ActorRef.noSender());
 374
 375         applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 376         verifyApplyState(applyState, leaderCollectorActor, "serverConfig", currentTerm, 8, serverConfig);
 377
 378         applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class);
 379         verifyApplyState(applyState, null, null, currentTerm, 8, serverConfig);
 380
 381         // Verify the leader's persisted snapshot.
 382         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 383         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
 384         verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 6, currentTerm, 7);
 385         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
 386         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 387         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 7, payload7);
 388
 389         expSnapshotState.add(payload7);
 390
 391         verifyInstallSnapshotToLaggingFollower(8, serverConfig);
 392
 393         testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot complete");
 394     }
 395
 396     /**
 397      * Tests whether the leader reattempts to send a snapshot when a follower crashes before replying with
 398      * InstallSnapshotReply after the last chunk has been sent.
 399      */
 400     @Test
 401     public void testLeaderInstallsSnapshotWithRestartedFollowerDuringSnapshotInstallation() throws Exception {
 402         testLog.info("testLeaderInstallsSnapshotWithRestartedFollowerDuringSnapshotInstallation starting");
 403
 404         setup();
 405
 406         sendInitialPayloadsReplicatedToAllFollowers("zero", "one");
 407
 408         // Configure follower 2 to drop messages and lag.
 409         follower2Actor.stop();
 410
 411         // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader.
 412         Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5,
 413                 TimeUnit.MILLISECONDS);
 414
 415         // Send 5 payloads - the second should cause a leader snapshot.
 416         final MockPayload payload2 = sendPayloadData(leaderActor, "two");
 417         final MockPayload payload3 = sendPayloadData(leaderActor, "three");
 418         final MockPayload payload4 = sendPayloadData(leaderActor, "four");
 419         final MockPayload payload5 = sendPayloadData(leaderActor, "five");
 420         final MockPayload payload6 = sendPayloadData(leaderActor, "six");
 421
 422         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 423
 424         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 425         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 5);
 426         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2);
 427         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
 428         verifyApplyState(applyStates.get(4), leaderCollectorActor, payload6.toString(), currentTerm, 6, payload6);
 429
 430         MessageCollectorActor.clearMessages(leaderCollectorActor);
 431
 432         testLog.info("testLeaderInstallsSnapshotWithRestartedFollowerDuringSnapshotInstallation: "
 433                 + "sending 1 more payload to trigger second snapshot");
 434
 435         // Send another payload to trigger a second leader snapshot.
 436         MockPayload payload7 = sendPayloadData(leaderActor, "seven");
 437
 438         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 439
 440
 441         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 442         verifyApplyState(applyState, leaderCollectorActor, payload7.toString(), currentTerm, 7, payload7);
 443
 444         // Verify follower 1 applies each log entry.
 445         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 6);
 446         verifyApplyState(applyStates.get(0), null, null, currentTerm, 2, payload2);
 447         verifyApplyState(applyStates.get(2), null, null, currentTerm, 4, payload4);
 448         verifyApplyState(applyStates.get(5), null, null, currentTerm, 7, payload7);
 449
 450         leaderActor.underlyingActor()
 451                 .startDropMessages(InstallSnapshotReply.class, reply -> reply.getChunkIndex() == 5);
 452
 453         setupFollower2();
 454
 455         MessageCollectorActor.expectMatching(follower2CollectorActor, InstallSnapshot.class, 5);
 456
 457         follower2Actor.stop();
 458
 459         // need to get rid of persistence for follower2
 460         InMemorySnapshotStore.clearSnapshotsFor(follower2Id);
 461
 462         leaderActor.underlyingActor().stopDropMessages(InstallSnapshotReply.class);
 463
 464         MessageCollectorActor.clearMessages(follower2CollectorActor);
 465         setupFollower2();
 466
 467         MessageCollectorActor.expectMatching(follower2CollectorActor, SaveSnapshotSuccess.class, 1);
 468     }
 469
 470     /**
 471      * Send payloads with follower 2 lagging with the last payload having a large enough size to trigger a
 472      * leader snapshot such that the leader trims its log from the last applied index.. Follower 2's log will
 473      * be behind by several entries and, when it is resumed, it should be caught up via a snapshot installed
 474      * by the leader.
 475      */
 476     @Test
 477     public void testLeaderSnapshotTriggeredByMemoryThresholdExceededWithLaggingFollower() {
 478         testLog.info("testLeaderSnapshotTriggeredByMemoryThresholdExceededWithLaggingFollower starting");
 479
 480         snapshotBatchCount = 5;
 481         setup();
 482
 483         sendInitialPayloadsReplicatedToAllFollowers("zero");
 484
 485         leaderActor.underlyingActor().setMockTotalMemory(1000);
 486
 487         // We'll expect a ReplicatedLogImplEntry message and an ApplyJournalEntries message added to the journal.
 488         InMemoryJournal.addWriteMessagesCompleteLatch(leaderId, 2);
 489
 490         follower2Actor.underlyingActor().startDropMessages(AppendEntries.class);
 491
 492         // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader.
 493         Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5,
 494                 TimeUnit.MILLISECONDS);
 495
 496         // Send a payload with a large relative size but not enough to trigger a snapshot.
 497         MockPayload payload1 = sendPayloadData(leaderActor, "one", 500);
 498
 499         // Verify the leader got consensus and applies the first log entry even though follower 2 didn't respond.
 500         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 1);
 501         verifyApplyState(applyStates.get(0), leaderCollectorActor, payload1.toString(), currentTerm, 1, payload1);
 502
 503         // Wait for all the ReplicatedLogImplEntry and ApplyJournalEntries messages to be added to the journal
 504         // before the snapshot so the snapshot sequence # will be higher to ensure the snapshot gets
 505         // purged from the snapshot store after subsequent snapshots.
 506         InMemoryJournal.waitForWriteMessagesComplete(leaderId);
 507
 508         // Verify a snapshot is not triggered.
 509         CaptureSnapshot captureSnapshot = MessageCollectorActor.getFirstMatching(leaderCollectorActor,
 510                 CaptureSnapshot.class);
 511         Assert.assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot);
 512
 513         expSnapshotState.add(payload1);
 514
 515         // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader.
 516         Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5,
 517                 TimeUnit.MILLISECONDS);
 518
 519         // Send another payload with a large enough relative size in combination with the last payload
 520         // that exceeds the memory threshold (70% * 1000 = 700) - this should do a snapshot.
 521         MockPayload payload2 = sendPayloadData(leaderActor, "two", 201);
 522
 523         // Verify the leader applies the last log entry.
 524         applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 2);
 525         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2);
 526
 527         // Verify follower 1 applies each log entry.
 528         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 2);
 529         verifyApplyState(applyStates.get(0), null, null, currentTerm, 1, payload1);
 530         verifyApplyState(applyStates.get(1), null, null, currentTerm, 2, payload2);
 531
 532         // A snapshot should've occurred - wait for it to complete.
 533         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 534
 535         // Because the snapshot was triggered by exceeding the memory threshold the leader should've advanced
 536         // the snapshot index to the last applied index and trimmed the log even though the entries weren't
 537         // replicated to all followers.
 538         verifyLeadersTrimmedLog(2, 0);
 539
 540         // Verify the leader's persisted snapshot.
 541         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 542         assertEquals("Persisted snapshots size", 1, persistedSnapshots.size());
 543         verifySnapshot("Persisted", persistedSnapshots.get(0), currentTerm, 1, currentTerm, 2);
 544         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshots.get(0).getUnAppliedEntries();
 545         assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 546         verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 2, payload2);
 547
 548         expSnapshotState.add(payload2);
 549
 550         verifyInstallSnapshotToLaggingFollower(2L, null);
 551
 552         // Sends a payload with index 3.
 553         verifyNoSubsequentSnapshotAfterMemoryThresholdExceededSnapshot();
 554
 555         // Sends 3 payloads with indexes 4, 5 and 6.
 556         long leadersSnapshotIndexOnRecovery = verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot();
 557
 558         // Recover the leader from persistence and verify.
 559         long leadersLastIndexOnRecovery = 6;
 560
 561         long leadersFirstJournalEntryIndexOnRecovery = leadersSnapshotIndexOnRecovery + 1;
 562
 563         verifyLeaderRecoveryAfterReinstatement(leadersLastIndexOnRecovery, leadersSnapshotIndexOnRecovery,
 564                 leadersFirstJournalEntryIndexOnRecovery);
 565
 566         testLog.info("testLeaderSnapshotTriggeredByMemoryThresholdExceeded ending");
 567     }
 568
 569     /**
 570      * Send another payload to verify another snapshot is not done since the last snapshot trimmed the
 571      * first log entry so the memory threshold should not be exceeded.
 572      */
 573     private void verifyNoSubsequentSnapshotAfterMemoryThresholdExceededSnapshot() {
 574         ApplyState applyState;
 575         CaptureSnapshot captureSnapshot;
 576
 577         MockPayload payload3 = sendPayloadData(leaderActor, "three");
 578
 579         // Verify the leader applies the state.
 580         applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 581         verifyApplyState(applyState, leaderCollectorActor, payload3.toString(), currentTerm, 3, payload3);
 582
 583         captureSnapshot = MessageCollectorActor.getFirstMatching(leaderCollectorActor, CaptureSnapshot.class);
 584         Assert.assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot);
 585
 586         // Verify the follower 1 applies the state.
 587         applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class);
 588         verifyApplyState(applyState, null, null, currentTerm, 3, payload3);
 589
 590         // Verify the follower 2 applies the state.
 591         applyState = MessageCollectorActor.expectFirstMatching(follower2CollectorActor, ApplyState.class);
 592         verifyApplyState(applyState, null, null, currentTerm, 3, payload3);
 593
 594         // Verify the leader's state.
 595         verifyLeadersTrimmedLog(3);
 596
 597         // Verify follower 1's state.
 598         verifyFollowersTrimmedLog(1, follower1Actor, 3);
 599
 600         // Verify follower 2's state.
 601         verifyFollowersTrimmedLog(2, follower2Actor, 3);
 602
 603         // Revert back to JVM total memory.
 604         leaderActor.underlyingActor().setMockTotalMemory(0);
 605
 606         MessageCollectorActor.clearMessages(leaderCollectorActor);
 607         MessageCollectorActor.clearMessages(follower1CollectorActor);
 608         MessageCollectorActor.clearMessages(follower2CollectorActor);
 609
 610         expSnapshotState.add(payload3);
 611     }
 612
 613     /**
 614      * Resume the lagging follower 2 and verify it receives an install snapshot from the leader.
 615      */
 616     private void verifyInstallSnapshotToLaggingFollower(long lastAppliedIndex,
 617             @Nullable ServerConfigurationPayload expServerConfig) {
 618         testLog.info("verifyInstallSnapshotToLaggingFollower starting");
 619
 620         MessageCollectorActor.clearMessages(leaderCollectorActor);
 621
 622         // Now stop dropping AppendEntries in follower 2.
 623         follower2Actor.underlyingActor().stopDropMessages(AppendEntries.class);
 624
 625
 626         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 627
 628         // Verify the leader's persisted snapshot. The previous snapshot (currently) won't be deleted from
 629         // the snapshot store because the second snapshot was initiated by the follower install snapshot and
 630         // not because the batch count was reached so the persisted journal sequence number wasn't advanced
 631         // far enough to cause the previous snapshot to be deleted. This is because
 632         // RaftActor#trimPersistentData subtracts the snapshotBatchCount from the snapshot's sequence number.
 633         // This is OK - the next snapshot should delete it. In production, even if the system restarted
 634         // before another snapshot, they would both get applied which wouldn't hurt anything.
 635         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 636         Assert.assertTrue("Expected at least 1 persisted snapshots", persistedSnapshots.size() > 0);
 637         Snapshot persistedSnapshot = persistedSnapshots.get(persistedSnapshots.size() - 1);
 638         verifySnapshot("Persisted", persistedSnapshot, currentTerm, lastAppliedIndex, currentTerm, lastAppliedIndex);
 639         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshot.getUnAppliedEntries();
 640         assertEquals("Persisted Snapshot getUnAppliedEntries size", 0, unAppliedEntry.size());
 641
 642         int snapshotSize = SerializationUtils.serialize(persistedSnapshot.getState()).length;
 643         final int expTotalChunks = snapshotSize / SNAPSHOT_CHUNK_SIZE
 644                 + (snapshotSize % SNAPSHOT_CHUNK_SIZE > 0 ? 1 : 0);
 645
 646         InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(follower2CollectorActor,
 647                 InstallSnapshot.class);
 648         assertEquals("InstallSnapshot getTerm", currentTerm, installSnapshot.getTerm());
 649         assertEquals("InstallSnapshot getLeaderId", leaderId, installSnapshot.getLeaderId());
 650         assertEquals("InstallSnapshot getChunkIndex", 1, installSnapshot.getChunkIndex());
 651         assertEquals("InstallSnapshot getTotalChunks", expTotalChunks, installSnapshot.getTotalChunks());
 652         assertEquals("InstallSnapshot getLastIncludedTerm", currentTerm, installSnapshot.getLastIncludedTerm());
 653         assertEquals("InstallSnapshot getLastIncludedIndex", lastAppliedIndex, installSnapshot.getLastIncludedIndex());
 654         //assertArrayEquals("InstallSnapshot getData", snapshot, installSnapshot.getData().toByteArray());
 655
 656         List<InstallSnapshotReply> installSnapshotReplies = MessageCollectorActor.expectMatching(
 657                 leaderCollectorActor, InstallSnapshotReply.class, expTotalChunks);
 658         int index = 1;
 659         for (InstallSnapshotReply installSnapshotReply: installSnapshotReplies) {
 660             assertEquals("InstallSnapshotReply getTerm", currentTerm, installSnapshotReply.getTerm());
 661             assertEquals("InstallSnapshotReply getChunkIndex", index++, installSnapshotReply.getChunkIndex());
 662             assertEquals("InstallSnapshotReply getFollowerId", follower2Id, installSnapshotReply.getFollowerId());
 663             assertEquals("InstallSnapshotReply isSuccess", true, installSnapshotReply.isSuccess());
 664         }
 665
 666         // Verify follower 2 applies the snapshot.
 667         ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(follower2CollectorActor,
 668                 ApplySnapshot.class);
 669         verifySnapshot("Follower 2", applySnapshot.getSnapshot(), currentTerm, lastAppliedIndex, currentTerm,
 670                 lastAppliedIndex);
 671         assertEquals("Persisted Snapshot getUnAppliedEntries size", 0,
 672                 applySnapshot.getSnapshot().getUnAppliedEntries().size());
 673
 674         // Wait for the snapshot to complete.
 675         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 676
 677         // Ensure there's at least 1 more heartbeat.
 678         MessageCollectorActor.clearMessages(leaderCollectorActor);
 679         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 680
 681         // The leader should now have performed fake snapshots to advance the snapshot index and to trim
 682         // the log. In addition replicatedToAllIndex should've advanced.
 683         verifyLeadersTrimmedLog(lastAppliedIndex);
 684
 685         if (expServerConfig != null) {
 686             Set<ServerInfo> expServerInfo = new HashSet<>(expServerConfig.getServerConfig());
 687             assertEquals("Leader snapshot server config", expServerInfo,
 688                     new HashSet<>(persistedSnapshot.getServerConfiguration().getServerConfig()));
 689
 690             assertEquals("Follower 2 snapshot server config", expServerInfo,
 691                     new HashSet<>(applySnapshot.getSnapshot().getServerConfiguration().getServerConfig()));
 692
 693             ServerConfigurationPayload follower2ServerConfig = follower2Context.getPeerServerInfo(true);
 694             assertNotNull("Follower 2 server config is null", follower2ServerConfig);
 695
 696             assertEquals("Follower 2 server config", expServerInfo,
 697                     new HashSet<>(follower2ServerConfig.getServerConfig()));
 698         }
 699
 700         MessageCollectorActor.clearMessages(leaderCollectorActor);
 701         MessageCollectorActor.clearMessages(follower1CollectorActor);
 702         MessageCollectorActor.clearMessages(follower2CollectorActor);
 703
 704         testLog.info("verifyInstallSnapshotToLaggingFollower complete");
 705     }
 706
 707     /**
 708      * Do another round of payloads and snapshot to verify replicatedToAllIndex gets back on track and
 709      * snapshots works as expected after doing a follower snapshot. In this step we don't lag a follower.
 710      */
 711     private long verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot() {
 712         testLog.info(
 713                 "verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot starting: replicatedToAllIndex: {}",
 714                 leader.getReplicatedToAllIndex());
 715
 716         // Send another payload - a snapshot should occur.
 717         MockPayload payload4 = sendPayloadData(leaderActor, "four");
 718
 719         // Wait for the snapshot to complete.
 720         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class);
 721
 722         ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class);
 723         verifyApplyState(applyState, leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4);
 724
 725         // Verify the leader's last persisted snapshot (previous ones may not be purged yet).
 726         List<Snapshot> persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class);
 727         Snapshot persistedSnapshot = persistedSnapshots.get(persistedSnapshots.size() - 1);
 728         // The last (fourth) payload may or may not have been applied when the snapshot is captured depending on the
 729         // timing when the async persistence completes.
 730         List<ReplicatedLogEntry> unAppliedEntry = persistedSnapshot.getUnAppliedEntries();
 731         long leadersSnapshotIndex;
 732         if (unAppliedEntry.isEmpty()) {
 733             leadersSnapshotIndex = 4;
 734             expSnapshotState.add(payload4);
 735             verifySnapshot("Persisted", persistedSnapshot, currentTerm, 4, currentTerm, 4);
 736         } else {
 737             leadersSnapshotIndex = 3;
 738             verifySnapshot("Persisted", persistedSnapshot, currentTerm, 3, currentTerm, 4);
 739             assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size());
 740             verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 4, payload4);
 741             expSnapshotState.add(payload4);
 742         }
 743
 744         // Send a couple more payloads.
 745         MockPayload payload5 = sendPayloadData(leaderActor, "five");
 746         MockPayload payload6 = sendPayloadData(leaderActor, "six");
 747
 748         // Verify the leader applies the 2 log entries.
 749         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 3);
 750         verifyApplyState(applyStates.get(1), leaderCollectorActor, payload5.toString(), currentTerm, 5, payload5);
 751         verifyApplyState(applyStates.get(2), leaderCollectorActor, payload6.toString(), currentTerm, 6, payload6);
 752
 753         // Verify the leader applies a log entry for at least the last entry index.
 754         verifyApplyJournalEntries(leaderCollectorActor, 6);
 755
 756         // Ensure there's at least 1 more heartbeat to trim the log.
 757         MessageCollectorActor.clearMessages(leaderCollectorActor);
 758         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 759
 760         // Verify the leader's final state.
 761         verifyLeadersTrimmedLog(6);
 762
 763         InMemoryJournal.dumpJournal(leaderId);
 764
 765         // Verify the leaders's persisted journal log - it should only contain the last 2 ReplicatedLogEntries
 766         // added after the snapshot as the persisted journal should've been purged to the snapshot
 767         // sequence number.
 768         verifyPersistedJournal(leaderId, Arrays.asList(new SimpleReplicatedLogEntry(5, currentTerm, payload5),
 769                 new SimpleReplicatedLogEntry(6, currentTerm, payload6)));
 770
 771         // Verify the leaders's persisted journal contains an ApplyJournalEntries for at least the last entry index.
 772         List<ApplyJournalEntries> persistedApplyJournalEntries =
 773                 InMemoryJournal.get(leaderId, ApplyJournalEntries.class);
 774         boolean found = false;
 775         for (ApplyJournalEntries entry: persistedApplyJournalEntries) {
 776             if (entry.getToIndex() == 6) {
 777                 found = true;
 778                 break;
 779             }
 780         }
 781
 782         Assert.assertTrue(String.format("ApplyJournalEntries with index %d not found in leader's persisted journal", 6),
 783                 found);
 784
 785         // Verify follower 1 applies the 3 log entries.
 786         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 3);
 787         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
 788         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
 789         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
 790
 791         // Verify follower 1's log state.
 792         verifyFollowersTrimmedLog(1, follower1Actor, 6);
 793
 794         // Verify follower 2 applies the 3 log entries.
 795         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, 3);
 796         verifyApplyState(applyStates.get(0), null, null, currentTerm, 4, payload4);
 797         verifyApplyState(applyStates.get(1), null, null, currentTerm, 5, payload5);
 798         verifyApplyState(applyStates.get(2), null, null, currentTerm, 6, payload6);
 799
 800         // Verify follower 2's log state.
 801         verifyFollowersTrimmedLog(2, follower2Actor, 6);
 802
 803         expSnapshotState.add(payload5);
 804         expSnapshotState.add(payload6);
 805
 806         testLog.info("verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot ending");
 807
 808         return leadersSnapshotIndex;
 809     }
 810
 811     /**
 812      * Kill the leader actor, reinstate it and verify the recovered journal.
 813      */
 814     private void verifyLeaderRecoveryAfterReinstatement(long lastIndex, long snapshotIndex,
 815             long firstJournalEntryIndex) {
 816         testLog.info("verifyLeaderRecoveryAfterReinstatement starting: lastIndex: {}, snapshotIndex: {}, "
 817             + "firstJournalEntryIndex: {}", lastIndex, snapshotIndex, firstJournalEntryIndex);
 818
 819         killActor(leaderActor);
 820
 821         leaderActor = newTestRaftActor(leaderId, peerAddresses, leaderConfigParams);
 822         TestRaftActor testRaftActor = leaderActor.underlyingActor();
 823
 824         testRaftActor.startDropMessages(RequestVoteReply.class);
 825
 826         leaderContext = testRaftActor.getRaftActorContext();
 827
 828         testRaftActor.waitForRecoveryComplete();
 829
 830         int logSize = (int) (expSnapshotState.size() - firstJournalEntryIndex);
 831         assertEquals("Leader snapshot term", currentTerm, leaderContext.getReplicatedLog().getSnapshotTerm());
 832         assertEquals("Leader snapshot index", snapshotIndex, leaderContext.getReplicatedLog().getSnapshotIndex());
 833         assertEquals("Leader journal log size", logSize, leaderContext.getReplicatedLog().size());
 834         assertEquals("Leader journal last index", lastIndex, leaderContext.getReplicatedLog().lastIndex());
 835         assertEquals("Leader commit index", lastIndex, leaderContext.getCommitIndex());
 836         assertEquals("Leader last applied", lastIndex, leaderContext.getLastApplied());
 837
 838         for (long i = firstJournalEntryIndex; i < expSnapshotState.size(); i++) {
 839             verifyReplicatedLogEntry(leaderContext.getReplicatedLog().get(i), currentTerm, i,
 840                     expSnapshotState.get((int) i));
 841         }
 842
 843         assertEquals("Leader applied state", expSnapshotState, testRaftActor.getState());
 844
 845         testLog.info("verifyLeaderRecoveryAfterReinstatement ending");
 846     }
 847
 848     private void sendInitialPayloadsReplicatedToAllFollowers(String... data) {
 849
 850         // Send the payloads.
 851         for (String d: data) {
 852             expSnapshotState.add(sendPayloadData(leaderActor, d));
 853         }
 854
 855         int numEntries = data.length;
 856
 857         // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond.
 858         List<ApplyState> applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor,
 859                 ApplyState.class, numEntries);
 860         for (int i = 0; i < expSnapshotState.size(); i++) {
 861             MockPayload payload = expSnapshotState.get(i);
 862             verifyApplyState(applyStates.get(i), leaderCollectorActor, payload.toString(), currentTerm, i, payload);
 863         }
 864
 865         // Verify follower 1 applies each log entry.
 866         applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, numEntries);
 867         for (int i = 0; i < expSnapshotState.size(); i++) {
 868             MockPayload payload = expSnapshotState.get(i);
 869             verifyApplyState(applyStates.get(i), null, null, currentTerm, i, payload);
 870         }
 871
 872         // Verify follower 2 applies each log entry.
 873         applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, numEntries);
 874         for (int i = 0; i < expSnapshotState.size(); i++) {
 875             MockPayload payload = expSnapshotState.get(i);
 876             verifyApplyState(applyStates.get(i), null, null, currentTerm, i, payload);
 877         }
 878
 879         // Ensure there's at least 1 more heartbeat.
 880         MessageCollectorActor.clearMessages(leaderCollectorActor);
 881         MessageCollectorActor.expectFirstMatching(leaderCollectorActor, AppendEntriesReply.class);
 882
 883         // The leader should have performed fake snapshots to trim the log to the last index replicated to
 884         // all followers.
 885         verifyLeadersTrimmedLog(numEntries - 1);
 886
 887         MessageCollectorActor.clearMessages(leaderCollectorActor);
 888         MessageCollectorActor.clearMessages(follower1CollectorActor);
 889         MessageCollectorActor.clearMessages(follower2CollectorActor);
 890     }
 891 }