X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FReplicationAndSnapshotsWithLaggingFollowerIntegrationTest.java;h=f2658957e1e173d66e30839cd3938ade1c57f870;hb=HEAD;hp=4211119e12e998a956e54feeecbaf323d96e2d8d;hpb=e1eca73a5ae2ffae8dd78c6fe5281cd2f45d5ef3;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest.java index 4211119e12..f2658957e1 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest.java @@ -8,20 +8,20 @@ package org.opendaylight.controller.cluster.raft; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import akka.actor.ActorRef; import akka.persistence.SaveSnapshotSuccess; -import com.google.common.collect.ImmutableMap; import com.google.common.util.concurrent.Uninterruptibles; -import java.util.Arrays; -import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; -import javax.annotation.Nullable; -import org.junit.Assert; +import org.apache.commons.lang3.SerializationUtils; +import org.eclipse.jdt.annotation.Nullable; import org.junit.Test; import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; @@ -36,6 +36,8 @@ import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries; import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload; import org.opendaylight.controller.cluster.raft.persisted.ServerInfo; +import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.persisted.Snapshot; import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore; @@ -58,15 +60,15 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A InMemoryJournal.addEntry(leaderId, 1, new UpdateElectionTerm(initialTerm, leaderId)); // Create the leader and 2 follower actors. - follower1Actor = newTestRaftActor(follower1Id, ImmutableMap.of(leaderId, testActorPath(leaderId), + follower1Actor = newTestRaftActor(follower1Id, Map.of(leaderId, testActorPath(leaderId), follower2Id, testActorPath(follower2Id)), newFollowerConfigParams()); - follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId), + follower2Actor = newTestRaftActor(follower2Id, Map.of(leaderId, testActorPath(leaderId), follower1Id, testActorPath(follower1Id)), newFollowerConfigParams()); - Map leaderPeerAddresses = ImmutableMap.builder() - .put(follower1Id, follower1Actor.path().toString()) - .put(follower2Id, follower2Actor.path().toString()).build(); + Map leaderPeerAddresses = Map.of( + follower1Id, follower1Actor.path().toString(), + follower2Id, follower2Actor.path().toString()); leaderConfigParams = newLeaderConfigParams(); leaderActor = newTestRaftActor(leaderId, leaderPeerAddresses, leaderConfigParams); @@ -83,7 +85,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A follower2 = follower2Actor.underlyingActor().getCurrentBehavior(); currentTerm = leaderContext.getTermInformation().getCurrentTerm(); - assertEquals("Current term > " + initialTerm, true, currentTerm > initialTerm); + assertTrue("Current term > " + initialTerm, currentTerm > initialTerm); leaderCollectorActor = leaderActor.underlyingActor().collectorActor(); follower1CollectorActor = follower1Actor.underlyingActor().collectorActor(); @@ -92,12 +94,22 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A testLog.info("Leader created and elected"); } + private void setupFollower2() { + follower2Actor = newTestRaftActor(follower2Id, Map.of(leaderId, testActorPath(leaderId), + follower1Id, testActorPath(follower1Id)), newFollowerConfigParams()); + + follower2Context = follower2Actor.underlyingActor().getRaftActorContext(); + follower2 = follower2Actor.underlyingActor().getCurrentBehavior(); + + follower2CollectorActor = follower2Actor.underlyingActor().collectorActor(); + } + /** * Send 2 payload instances with follower 2 lagging then resume the follower and verifies it gets * caught up via AppendEntries. */ @Test - public void testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries() throws Exception { + public void testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries() { testLog.info("testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries starting: sending 2 new payloads"); setup(); @@ -156,7 +168,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // to catch it up because no snapshotting was done so the follower's next index was present in the log. InstallSnapshot installSnapshot = MessageCollectorActor.getFirstMatching(follower2CollectorActor, InstallSnapshot.class); - Assert.assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot); + assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot); testLog.info("testReplicationsWithLaggingFollowerCaughtUpViaAppendEntries complete"); } @@ -168,7 +180,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A * sent by the leader. */ @Test - public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries() throws Exception { + public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries() { testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaAppendEntries starting"); setup(); @@ -241,7 +253,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // Verify the leader did not try to install a snapshot to catch up follower 2. InstallSnapshot installSnapshot = MessageCollectorActor.getFirstMatching(follower2CollectorActor, InstallSnapshot.class); - Assert.assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot); + assertNull("Follower 2 received unexpected InstallSnapshot", installSnapshot); // Ensure there's at least 1 more heartbeat. MessageCollectorActor.clearMessages(leaderCollectorActor); @@ -287,7 +299,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A * installed by the leader. */ @Test - public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot() throws Exception { + public void testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot() { testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot starting"); setup(); @@ -297,6 +309,10 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // Configure follower 2 to drop messages and lag. follower2Actor.underlyingActor().startDropMessages(AppendEntries.class); + // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader. + Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5, + TimeUnit.MILLISECONDS); + // Send 5 payloads - the second should cause a leader snapshot. final MockPayload payload2 = sendPayloadData(leaderActor, "two"); final MockPayload payload3 = sendPayloadData(leaderActor, "three"); @@ -317,10 +333,6 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot: " + "sending 1 more payload to trigger second snapshot"); - // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader. - Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5, - TimeUnit.MILLISECONDS); - // Send another payload to trigger a second leader snapshot. MockPayload payload7 = sendPayloadData(leaderActor, "seven"); @@ -351,7 +363,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // Send a server config change to test that the install snapshot includes the server config. - ServerConfigurationPayload serverConfig = new ServerConfigurationPayload(Arrays.asList( + ServerConfigurationPayload serverConfig = new ServerConfigurationPayload(List.of( new ServerInfo(leaderId, true), new ServerInfo(follower1Id, false), new ServerInfo(follower2Id, false))); @@ -380,6 +392,80 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A testLog.info("testLeaderSnapshotWithLaggingFollowerCaughtUpViaInstallSnapshot complete"); } + /** + * Tests whether the leader reattempts to send a snapshot when a follower crashes before replying with + * InstallSnapshotReply after the last chunk has been sent. + */ + @Test + public void testLeaderInstallsSnapshotWithRestartedFollowerDuringSnapshotInstallation() throws Exception { + testLog.info("testLeaderInstallsSnapshotWithRestartedFollowerDuringSnapshotInstallation starting"); + + setup(); + + sendInitialPayloadsReplicatedToAllFollowers("zero", "one"); + + // Configure follower 2 to drop messages and lag. + follower2Actor.stop(); + + // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader. + Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5, + TimeUnit.MILLISECONDS); + + // Send 5 payloads - the second should cause a leader snapshot. + final MockPayload payload2 = sendPayloadData(leaderActor, "two"); + final MockPayload payload3 = sendPayloadData(leaderActor, "three"); + final MockPayload payload4 = sendPayloadData(leaderActor, "four"); + final MockPayload payload5 = sendPayloadData(leaderActor, "five"); + final MockPayload payload6 = sendPayloadData(leaderActor, "six"); + + MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class); + + // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond. + List applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, ApplyState.class, 5); + verifyApplyState(applyStates.get(0), leaderCollectorActor, payload2.toString(), currentTerm, 2, payload2); + verifyApplyState(applyStates.get(2), leaderCollectorActor, payload4.toString(), currentTerm, 4, payload4); + verifyApplyState(applyStates.get(4), leaderCollectorActor, payload6.toString(), currentTerm, 6, payload6); + + MessageCollectorActor.clearMessages(leaderCollectorActor); + + testLog.info("testLeaderInstallsSnapshotWithRestartedFollowerDuringSnapshotInstallation: " + + "sending 1 more payload to trigger second snapshot"); + + // Send another payload to trigger a second leader snapshot. + MockPayload payload7 = sendPayloadData(leaderActor, "seven"); + + MessageCollectorActor.expectFirstMatching(leaderCollectorActor, SaveSnapshotSuccess.class); + + + ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollectorActor, ApplyState.class); + verifyApplyState(applyState, leaderCollectorActor, payload7.toString(), currentTerm, 7, payload7); + + // Verify follower 1 applies each log entry. + applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 6); + verifyApplyState(applyStates.get(0), null, null, currentTerm, 2, payload2); + verifyApplyState(applyStates.get(2), null, null, currentTerm, 4, payload4); + verifyApplyState(applyStates.get(5), null, null, currentTerm, 7, payload7); + + leaderActor.underlyingActor() + .startDropMessages(InstallSnapshotReply.class, reply -> reply.getChunkIndex() == 5); + + setupFollower2(); + + MessageCollectorActor.expectMatching(follower2CollectorActor, InstallSnapshot.class, 1); + + follower2Actor.stop(); + + // need to get rid of persistence for follower2 + InMemorySnapshotStore.clearSnapshotsFor(follower2Id); + + leaderActor.underlyingActor().stopDropMessages(InstallSnapshotReply.class); + + MessageCollectorActor.clearMessages(follower2CollectorActor); + setupFollower2(); + + MessageCollectorActor.expectMatching(follower2CollectorActor, SaveSnapshotSuccess.class, 1); + } + /** * Send payloads with follower 2 lagging with the last payload having a large enough size to trigger a * leader snapshot such that the leader trims its log from the last applied index.. Follower 2's log will @@ -387,7 +473,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A * by the leader. */ @Test - public void testLeaderSnapshotTriggeredByMemoryThresholdExceededWithLaggingFollower() throws Exception { + public void testLeaderSnapshotTriggeredByMemoryThresholdExceededWithLaggingFollower() { testLog.info("testLeaderSnapshotTriggeredByMemoryThresholdExceededWithLaggingFollower starting"); snapshotBatchCount = 5; @@ -402,6 +488,10 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A follower2Actor.underlyingActor().startDropMessages(AppendEntries.class); + // Sleep for at least the election timeout interval so follower 2 is deemed inactive by the leader. + Uninterruptibles.sleepUninterruptibly(leaderConfigParams.getElectionTimeOutInterval().toMillis() + 5, + TimeUnit.MILLISECONDS); + // Send a payload with a large relative size but not enough to trigger a snapshot. MockPayload payload1 = sendPayloadData(leaderActor, "one", 500); @@ -417,7 +507,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // Verify a snapshot is not triggered. CaptureSnapshot captureSnapshot = MessageCollectorActor.getFirstMatching(leaderCollectorActor, CaptureSnapshot.class); - Assert.assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot); + assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot); expSnapshotState.add(payload1); @@ -462,16 +552,12 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A verifyNoSubsequentSnapshotAfterMemoryThresholdExceededSnapshot(); // Sends 3 payloads with indexes 4, 5 and 6. - verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot(); + long leadersSnapshotIndexOnRecovery = verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot(); // Recover the leader from persistence and verify. long leadersLastIndexOnRecovery = 6; - // The leader's last snapshot was triggered by index 4 so the last applied index in the snapshot was 3. - long leadersSnapshotIndexOnRecovery = 3; - - // The recovered journal should have 3 entries starting at index 4. - long leadersFirstJournalEntryIndexOnRecovery = 4; + long leadersFirstJournalEntryIndexOnRecovery = leadersSnapshotIndexOnRecovery + 1; verifyLeaderRecoveryAfterReinstatement(leadersLastIndexOnRecovery, leadersSnapshotIndexOnRecovery, leadersFirstJournalEntryIndexOnRecovery); @@ -483,7 +569,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A * Send another payload to verify another snapshot is not done since the last snapshot trimmed the * first log entry so the memory threshold should not be exceeded. */ - private void verifyNoSubsequentSnapshotAfterMemoryThresholdExceededSnapshot() throws Exception { + private void verifyNoSubsequentSnapshotAfterMemoryThresholdExceededSnapshot() { ApplyState applyState; CaptureSnapshot captureSnapshot; @@ -494,7 +580,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A verifyApplyState(applyState, leaderCollectorActor, payload3.toString(), currentTerm, 3, payload3); captureSnapshot = MessageCollectorActor.getFirstMatching(leaderCollectorActor, CaptureSnapshot.class); - Assert.assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot); + assertNull("Leader received unexpected CaptureSnapshot", captureSnapshot); // Verify the follower 1 applies the state. applyState = MessageCollectorActor.expectFirstMatching(follower1CollectorActor, ApplyState.class); @@ -526,9 +612,9 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A /** * Resume the lagging follower 2 and verify it receives an install snapshot from the leader. */ - private void verifyInstallSnapshotToLaggingFollower(long lastAppliedIndex, - @Nullable ServerConfigurationPayload expServerConfig) throws Exception { - testLog.info("testInstallSnapshotToLaggingFollower starting"); + private void verifyInstallSnapshotToLaggingFollower(final long lastAppliedIndex, + final @Nullable ServerConfigurationPayload expServerConfig) { + testLog.info("verifyInstallSnapshotToLaggingFollower starting"); MessageCollectorActor.clearMessages(leaderCollectorActor); @@ -546,15 +632,15 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // This is OK - the next snapshot should delete it. In production, even if the system restarted // before another snapshot, they would both get applied which wouldn't hurt anything. List persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class); - Assert.assertTrue("Expected at least 1 persisted snapshots", persistedSnapshots.size() > 0); + assertFalse("Expected at least 1 persisted snapshots", persistedSnapshots.isEmpty()); Snapshot persistedSnapshot = persistedSnapshots.get(persistedSnapshots.size() - 1); verifySnapshot("Persisted", persistedSnapshot, currentTerm, lastAppliedIndex, currentTerm, lastAppliedIndex); List unAppliedEntry = persistedSnapshot.getUnAppliedEntries(); assertEquals("Persisted Snapshot getUnAppliedEntries size", 0, unAppliedEntry.size()); - int snapshotSize = persistedSnapshot.getState().length; - final int expTotalChunks = snapshotSize / SNAPSHOT_CHUNK_SIZE - + (snapshotSize % SNAPSHOT_CHUNK_SIZE > 0 ? 1 : 0); + int snapshotSize = SerializationUtils.serialize(persistedSnapshot.getState()).length; + final int expTotalChunks = snapshotSize / MAXIMUM_MESSAGE_SLICE_SIZE + + (snapshotSize % MAXIMUM_MESSAGE_SLICE_SIZE > 0 ? 1 : 0); InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(follower2CollectorActor, InstallSnapshot.class); @@ -573,7 +659,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A assertEquals("InstallSnapshotReply getTerm", currentTerm, installSnapshotReply.getTerm()); assertEquals("InstallSnapshotReply getChunkIndex", index++, installSnapshotReply.getChunkIndex()); assertEquals("InstallSnapshotReply getFollowerId", follower2Id, installSnapshotReply.getFollowerId()); - assertEquals("InstallSnapshotReply isSuccess", true, installSnapshotReply.isSuccess()); + assertTrue("InstallSnapshotReply isSuccess", installSnapshotReply.isSuccess()); } // Verify follower 2 applies the snapshot. @@ -596,33 +682,34 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A verifyLeadersTrimmedLog(lastAppliedIndex); if (expServerConfig != null) { - Set expServerInfo = new HashSet<>(expServerConfig.getServerConfig()); + Set expServerInfo = Set.copyOf(expServerConfig.getServerConfig()); assertEquals("Leader snapshot server config", expServerInfo, - new HashSet<>(persistedSnapshot.getServerConfiguration().getServerConfig())); + Set.copyOf(persistedSnapshot.getServerConfiguration().getServerConfig())); assertEquals("Follower 2 snapshot server config", expServerInfo, - new HashSet<>(applySnapshot.getSnapshot().getServerConfiguration().getServerConfig())); + Set.copyOf(applySnapshot.getSnapshot().getServerConfiguration().getServerConfig())); ServerConfigurationPayload follower2ServerConfig = follower2Context.getPeerServerInfo(true); assertNotNull("Follower 2 server config is null", follower2ServerConfig); assertEquals("Follower 2 server config", expServerInfo, - new HashSet<>(follower2ServerConfig.getServerConfig())); + Set.copyOf(follower2ServerConfig.getServerConfig())); } MessageCollectorActor.clearMessages(leaderCollectorActor); MessageCollectorActor.clearMessages(follower1CollectorActor); MessageCollectorActor.clearMessages(follower2CollectorActor); - testLog.info("testInstallSnapshotToLaggingFollower complete"); + testLog.info("verifyInstallSnapshotToLaggingFollower complete"); } /** * Do another round of payloads and snapshot to verify replicatedToAllIndex gets back on track and * snapshots works as expected after doing a follower snapshot. In this step we don't lag a follower. */ - private void verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot() throws Exception { - testLog.info("testReplicationsAndSnapshotAfterInstallSnapshot starting: replicatedToAllIndex: {}", + private long verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot() { + testLog.info( + "verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot starting: replicatedToAllIndex: {}", leader.getReplicatedToAllIndex()); // Send another payload - a snapshot should occur. @@ -637,10 +724,21 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // Verify the leader's last persisted snapshot (previous ones may not be purged yet). List persistedSnapshots = InMemorySnapshotStore.getSnapshots(leaderId, Snapshot.class); Snapshot persistedSnapshot = persistedSnapshots.get(persistedSnapshots.size() - 1); - verifySnapshot("Persisted", persistedSnapshot, currentTerm, 3, currentTerm, 4); + // The last (fourth) payload may or may not have been applied when the snapshot is captured depending on the + // timing when the async persistence completes. List unAppliedEntry = persistedSnapshot.getUnAppliedEntries(); - assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size()); - verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 4, payload4); + long leadersSnapshotIndex; + if (unAppliedEntry.isEmpty()) { + leadersSnapshotIndex = 4; + expSnapshotState.add(payload4); + verifySnapshot("Persisted", persistedSnapshot, currentTerm, 4, currentTerm, 4); + } else { + leadersSnapshotIndex = 3; + verifySnapshot("Persisted", persistedSnapshot, currentTerm, 3, currentTerm, 4); + assertEquals("Persisted Snapshot getUnAppliedEntries size", 1, unAppliedEntry.size()); + verifyReplicatedLogEntry(unAppliedEntry.get(0), currentTerm, 4, payload4); + expSnapshotState.add(payload4); + } // Send a couple more payloads. MockPayload payload5 = sendPayloadData(leaderActor, "five"); @@ -666,8 +764,9 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // Verify the leaders's persisted journal log - it should only contain the last 2 ReplicatedLogEntries // added after the snapshot as the persisted journal should've been purged to the snapshot // sequence number. - verifyPersistedJournal(leaderId, Arrays.asList(new ReplicatedLogImplEntry(5, currentTerm, payload5), - new ReplicatedLogImplEntry(6, currentTerm, payload6))); + verifyPersistedJournal(leaderId, List.of( + new SimpleReplicatedLogEntry(5, currentTerm, payload5), + new SimpleReplicatedLogEntry(6, currentTerm, payload6))); // Verify the leaders's persisted journal contains an ApplyJournalEntries for at least the last entry index. List persistedApplyJournalEntries = @@ -680,8 +779,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A } } - Assert.assertTrue(String.format("ApplyJournalEntries with index %d not found in leader's persisted journal", 6), - found); + assertTrue("ApplyJournalEntries with index 6 not found in leader's persisted journal", found); // Verify follower 1 applies the 3 log entries. applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, 3); @@ -701,19 +799,21 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A // Verify follower 2's log state. verifyFollowersTrimmedLog(2, follower2Actor, 6); - expSnapshotState.add(payload4); expSnapshotState.add(payload5); expSnapshotState.add(payload6); - testLog.info("testReplicationsAndSnapshotAfterInstallSnapshot ending"); + testLog.info("verifyReplicationsAndSnapshotWithNoLaggingAfterInstallSnapshot ending"); + + return leadersSnapshotIndex; } /** * Kill the leader actor, reinstate it and verify the recovered journal. */ - private void verifyLeaderRecoveryAfterReinstatement(long lastIndex, long snapshotIndex, - long firstJournalEntryIndex) { - testLog.info("testLeaderReinstatement starting"); + private void verifyLeaderRecoveryAfterReinstatement(final long lastIndex, final long snapshotIndex, + final long firstJournalEntryIndex) { + testLog.info("verifyLeaderRecoveryAfterReinstatement starting: lastIndex: {}, snapshotIndex: {}, " + + "firstJournalEntryIndex: {}", lastIndex, snapshotIndex, firstJournalEntryIndex); killActor(leaderActor); @@ -741,11 +841,10 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A assertEquals("Leader applied state", expSnapshotState, testRaftActor.getState()); - testLog.info("testLeaderReinstatement ending"); + testLog.info("verifyLeaderRecoveryAfterReinstatement ending"); } - private void sendInitialPayloadsReplicatedToAllFollowers(String... data) { - + private void sendInitialPayloadsReplicatedToAllFollowers(final String... data) { // Send the payloads. for (String d: data) { expSnapshotState.add(sendPayloadData(leaderActor, d)); @@ -754,25 +853,27 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A int numEntries = data.length; // Verify the leader got consensus and applies each log entry even though follower 2 didn't respond. - List applyStates = MessageCollectorActor.expectMatching(leaderCollectorActor, - ApplyState.class, numEntries); + final var leaderStates = MessageCollectorActor.expectMatching(leaderCollectorActor, + ApplyState.class, numEntries); for (int i = 0; i < expSnapshotState.size(); i++) { - MockPayload payload = expSnapshotState.get(i); - verifyApplyState(applyStates.get(i), leaderCollectorActor, payload.toString(), currentTerm, i, payload); + final MockPayload payload = expSnapshotState.get(i); + verifyApplyState(leaderStates.get(i), leaderCollectorActor, payload.toString(), currentTerm, i, payload); } // Verify follower 1 applies each log entry. - applyStates = MessageCollectorActor.expectMatching(follower1CollectorActor, ApplyState.class, numEntries); + final var follower1States = MessageCollectorActor.expectMatching(follower1CollectorActor, + ApplyState.class, numEntries); for (int i = 0; i < expSnapshotState.size(); i++) { - MockPayload payload = expSnapshotState.get(i); - verifyApplyState(applyStates.get(i), null, null, currentTerm, i, payload); + final MockPayload payload = expSnapshotState.get(i); + verifyApplyState(follower1States.get(i), null, null, currentTerm, i, payload); } // Verify follower 2 applies each log entry. - applyStates = MessageCollectorActor.expectMatching(follower2CollectorActor, ApplyState.class, numEntries); + final var follower2States = MessageCollectorActor.expectMatching(follower2CollectorActor, + ApplyState.class, numEntries); for (int i = 0; i < expSnapshotState.size(); i++) { - MockPayload payload = expSnapshotState.get(i); - verifyApplyState(applyStates.get(i), null, null, currentTerm, i, payload); + final MockPayload payload = expSnapshotState.get(i); + verifyApplyState(follower2States.get(i), null, null, currentTerm, i, payload); } // Ensure there's at least 1 more heartbeat.