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