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=00147a3c0e0588e19314cfcfe0be0c3fd931e62e;hpb=86e8e4a06b682aa772c834a2cef56d0596540e1b;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 00147a3c0e..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,21 +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 org.apache.commons.lang3.SerializationUtils; import org.eclipse.jdt.annotation.Nullable; -import org.junit.Assert; import org.junit.Test; import org.opendaylight.controller.cluster.raft.MockRaftActorContext.MockPayload; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; @@ -61,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); @@ -86,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(); @@ -96,7 +95,7 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A } private void setupFollower2() { - follower2Actor = newTestRaftActor(follower2Id, ImmutableMap.of(leaderId, testActorPath(leaderId), + follower2Actor = newTestRaftActor(follower2Id, Map.of(leaderId, testActorPath(leaderId), follower1Id, testActorPath(follower1Id)), newFollowerConfigParams()); follower2Context = follower2Actor.underlyingActor().getRaftActorContext(); @@ -169,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"); } @@ -254,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); @@ -364,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))); @@ -508,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); @@ -581,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); @@ -633,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 = SerializationUtils.serialize(persistedSnapshot.getState()).length; - final int expTotalChunks = snapshotSize / SNAPSHOT_CHUNK_SIZE - + (snapshotSize % SNAPSHOT_CHUNK_SIZE > 0 ? 1 : 0); + final int expTotalChunks = snapshotSize / MAXIMUM_MESSAGE_SLICE_SIZE + + (snapshotSize % MAXIMUM_MESSAGE_SLICE_SIZE > 0 ? 1 : 0); InstallSnapshot installSnapshot = MessageCollectorActor.expectFirstMatching(follower2CollectorActor, InstallSnapshot.class); @@ -660,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. @@ -683,18 +682,18 @@ 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); @@ -765,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 SimpleReplicatedLogEntry(5, currentTerm, payload5), - new SimpleReplicatedLogEntry(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 = @@ -779,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); @@ -846,7 +845,6 @@ public class ReplicationAndSnapshotsWithLaggingFollowerIntegrationTest extends A } private void sendInitialPayloadsReplicatedToAllFollowers(final String... data) { - // Send the payloads. for (String d: data) { expSnapshotState.add(sendPayloadData(leaderActor, d)); @@ -855,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.