X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActorServerConfigurationSupportTest.java;h=a19281ffd8cbedd86062c008edd65cc6c2bbb8f4;hp=3145950b80ed5ca5af4b4cac2bd110d83d9fa407;hb=913ae866cd0cc82991e1f66ac80f6a42b0daaa48;hpb=92cbb07ef81943b0740ba7c8915001ac6785f560 diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorServerConfigurationSupportTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorServerConfigurationSupportTest.java index 3145950b80..a19281ffd8 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorServerConfigurationSupportTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorServerConfigurationSupportTest.java @@ -14,6 +14,7 @@ import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorAct import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor.clearMessages; import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor.expectFirstMatching; import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor.expectMatching; + import akka.actor.ActorRef; import akka.actor.Props; import akka.actor.UntypedActor; @@ -59,6 +60,7 @@ import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSn 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.UpdateElectionTerm; import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy; import org.opendaylight.controller.cluster.raft.utils.ForwardMessageToBehaviorActor; @@ -103,6 +105,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { InMemorySnapshotStore.clear(); } + @SuppressWarnings("checkstyle:IllegalCatch") private void setupNewFollower() { DefaultConfigParamsImpl configParams = newFollowerConfigParams(); @@ -157,7 +160,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { clearMessages(followerActor); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); @@ -198,30 +201,23 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { assertEquals("Follower peers", Sets.newHashSet(LEADER_ID, NEW_SERVER_ID), followerActorContext.getPeerIds()); - assertEquals("New follower peers", Sets.newHashSet(LEADER_ID, FOLLOWER_ID), newFollowerActorContext.getPeerIds()); - - expectFirstMatching(newFollowerCollectorActor, ApplyState.class); - expectFirstMatching(followerActor, ApplyState.class); + assertEquals("New follower peers", Sets.newHashSet(LEADER_ID, FOLLOWER_ID), + newFollowerActorContext.getPeerIds()); assertEquals("Follower commit index", 3, followerActorContext.getCommitIndex()); assertEquals("Follower last applied index", 3, followerActorContext.getLastApplied()); assertEquals("New follower commit index", 3, newFollowerActorContext.getCommitIndex()); assertEquals("New follower last applied index", 3, newFollowerActorContext.getLastApplied()); - List persistedLogEntries = InMemoryJournal.get(LEADER_ID, ReplicatedLogImplEntry.class); - assertEquals("Leader ReplicatedLogImplEntry entries", 1, persistedLogEntries.size()); - ReplicatedLogImplEntry logEntry = persistedLogEntries.get(0); - assertEquals("Leader ReplicatedLogImplEntry getTerm", 1, logEntry.getTerm()); - assertEquals("Leader ReplicatedLogImplEntry getIndex", 3, logEntry.getIndex()); - assertEquals("Leader ReplicatedLogImplEntry getData", ServerConfigurationPayload.class, logEntry.getData().getClass()); - - persistedLogEntries = InMemoryJournal.get(NEW_SERVER_ID, ReplicatedLogImplEntry.class); - assertEquals("New follower ReplicatedLogImplEntry entries", 1, persistedLogEntries.size()); - logEntry = persistedLogEntries.get(0); - assertEquals("New follower ReplicatedLogImplEntry getTerm", 1, logEntry.getTerm()); - assertEquals("New follower ReplicatedLogImplEntry getIndex", 3, logEntry.getIndex()); - assertEquals("New follower ReplicatedLogImplEntry getData", ServerConfigurationPayload.class, - logEntry.getData().getClass()); + assertEquals("Leader persisted ReplicatedLogImplEntry entries", 0, + InMemoryJournal.get(LEADER_ID, SimpleReplicatedLogEntry.class).size()); + assertEquals("Leader persisted ServerConfigurationPayload entries", 1, + InMemoryJournal.get(LEADER_ID, ServerConfigurationPayload.class).size()); + + assertEquals("New follower persisted ReplicatedLogImplEntry entries", 0, + InMemoryJournal.get(NEW_SERVER_ID, SimpleReplicatedLogEntry.class).size()); + assertEquals("New follower persisted ServerConfigurationPayload entries", 1, + InMemoryJournal.get(NEW_SERVER_ID, ServerConfigurationPayload.class).size()); LOG.info("testAddServerWithExistingFollower ending"); } @@ -243,9 +239,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); - RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + final RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); @@ -296,9 +292,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); - RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + final RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), false), testKit.getRef()); @@ -366,9 +362,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); - RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + final RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); RaftActorContext follower2ActorContext = newFollowerContext(NEW_SERVER_ID2, followerActor); Follower newFollower2 = new Follower(follower2ActorContext); @@ -429,7 +425,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); - RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + final RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); @@ -511,7 +507,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(100); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop the commit message so the snapshot doesn't complete yet. leaderRaftActor.setDropMessageOfType(COMMIT_MESSAGE_CLASS); @@ -568,7 +564,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); - UnInitializedFollowerSnapshotReply snapshotReply = expectFirstMatching(leaderCollectorActor, + final UnInitializedFollowerSnapshotReply snapshotReply = expectFirstMatching(leaderCollectorActor, UnInitializedFollowerSnapshotReply.class); // Prevent election timeout when the leader switches to follower @@ -632,12 +628,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef noLeaderActor = actorFactory.createTestActor( MockRaftActor.builder().id(LEADER_ID).peerAddresses(ImmutableMap.of(FOLLOWER_ID, - followerActor.path().toString())).config(configParams).persistent(Optional.of(false)). - props().withDispatcher(Dispatchers.DefaultDispatcherId()), + followerActor.path().toString())).config(configParams).persistent(Optional.of(false)) + .props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); noLeaderActor.underlyingActor().waitForInitializeBehaviorComplete(); - noLeaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); + noLeaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), + testKit.getRef()); AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); @@ -657,16 +654,15 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); - RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + final RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop UnInitializedFollowerSnapshotReply initially leaderRaftActor.setDropMessageOfType(UnInitializedFollowerSnapshotReply.class); MockNewFollowerRaftActor newFollowerRaftActorInstance = newFollowerRaftActor.underlyingActor(); - TestActorRef newFollowerCollectorActor = - newCollectorActor(newFollowerRaftActorInstance, NEW_SERVER_ID); + newFollowerCollectorActor = newCollectorActor(newFollowerRaftActorInstance, NEW_SERVER_ID); // Drop AppendEntries to the new follower so consensus isn't reached newFollowerRaftActorInstance.setDropMessageOfType(AppendEntries.class); @@ -739,15 +735,16 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef followerRaftActor = actorFactory.createTestActor( MockRaftActor.builder().id(FOLLOWER_ID).peerAddresses(ImmutableMap.of(LEADER_ID, - leaderActor.path().toString())).config(configParams).persistent(Optional.of(false)). - props().withDispatcher(Dispatchers.DefaultDispatcherId()), + leaderActor.path().toString())).config(configParams).persistent(Optional.of(false)) + .props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(FOLLOWER_ID)); followerRaftActor.underlyingActor().waitForInitializeBehaviorComplete(); followerRaftActor.tell(new AppendEntries(1, LEADER_ID, 0, 1, Collections.emptyList(), -1, -1, (short)0), leaderActor); - followerRaftActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); + followerRaftActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), + testKit.getRef()); expectFirstMatching(leaderActor, AddServer.class); LOG.info("testAddServerForwardedToLeader ending"); @@ -761,18 +758,19 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); TestActorRef noLeaderActor = actorFactory.createTestActor( MockRaftActor.builder().id(LEADER_ID).peerAddresses(ImmutableMap.of(FOLLOWER_ID, - followerActor.path().toString())).config(configParams).persistent(Optional.of(false)). - props().withDispatcher(Dispatchers.DefaultDispatcherId()), + followerActor.path().toString())).config(configParams).persistent(Optional.of(false)) + .props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); - RaftActorServerConfigurationSupport support = new RaftActorServerConfigurationSupport(noLeaderActor.underlyingActor()); + RaftActorServerConfigurationSupport support = new RaftActorServerConfigurationSupport( + noLeaderActor.underlyingActor()); - ReplicatedLogEntry serverConfigEntry = new MockRaftActorContext.MockReplicatedLogEntry(1, 1, + ReplicatedLogEntry serverConfigEntry = new SimpleReplicatedLogEntry(1, 1, new ServerConfigurationPayload(Collections.emptyList())); boolean handled = support.handleMessage(new ApplyState(null, null, serverConfigEntry), ActorRef.noSender()); assertEquals("Message handled", true, handled); - ReplicatedLogEntry nonServerConfigEntry = new MockRaftActorContext.MockReplicatedLogEntry(1, 1, + ReplicatedLogEntry nonServerConfigEntry = new SimpleReplicatedLogEntry(1, 1, new MockRaftActorContext.MockPayload("1")); handled = support.handleMessage(new ApplyState(null, null, nonServerConfigEntry), ActorRef.noSender()); assertEquals("Message handled", false, handled); @@ -789,13 +787,14 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef leaderActor = actorFactory.createTestActor( MockRaftActor.builder().id(LEADER_ID).peerAddresses(ImmutableMap.of(FOLLOWER_ID, - followerActor.path().toString())).config(configParams).persistent(Optional.of(false)). - props().withDispatcher(Dispatchers.DefaultDispatcherId()), + followerActor.path().toString())).config(configParams).persistent(Optional.of(false)) + .props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); leaderActor.underlyingActor().waitForInitializeBehaviorComplete(); leaderActor.tell(new RemoveServer(FOLLOWER_ID), testKit.getRef()); - RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), + RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, removeServerReply.getStatus()); LOG.info("testRemoveServerWithNoLeader ending"); @@ -813,7 +812,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); leaderActor.tell(new RemoveServer(NEW_SERVER_ID), testKit.getRef()); - RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), + RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.DOES_NOT_EXIST, removeServerReply.getStatus()); LOG.info("testRemoveServerNonExistentServer ending"); @@ -832,8 +832,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef followerRaftActor = actorFactory.createTestActor( MockRaftActor.builder().id(FOLLOWER_ID).peerAddresses(ImmutableMap.of(LEADER_ID, - leaderActor.path().toString())).config(configParams).persistent(Optional.of(false)). - props().withDispatcher(Dispatchers.DefaultDispatcherId()), + leaderActor.path().toString())).config(configParams).persistent(Optional.of(false)) + .props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(FOLLOWER_ID)); followerRaftActor.underlyingActor().waitForInitializeBehaviorComplete(); @@ -863,23 +863,26 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); - TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + final TestActorRef leaderCollector = + newLeaderCollectorActor(leaderActor.underlyingActor()); - TestActorRef collector = - actorFactory.createTestActor(MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + TestActorRef collector = actorFactory.createTestActor(MessageCollectorActor.props() + .withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); - TestActorRef followerRaftActor = actorFactory.createTestActor( + actorFactory.createTestActor( CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), configParams, NO_PERSISTENCE, collector).withDispatcher(Dispatchers.DefaultDispatcherId()), followerActorId); leaderActor.tell(new RemoveServer(FOLLOWER_ID), testKit.getRef()); - RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), + RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, removeServerReply.getStatus()); final ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); assertEquals(0L, applyState.getReplicatedLogEntry().getIndex()); - verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), votingServer(LEADER_ID)); + verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + votingServer(LEADER_ID)); RaftActorBehavior currentBehavior = leaderActor.underlyingActor().getCurrentBehavior(); assertTrue("Expected Leader", currentBehavior instanceof Leader); @@ -907,17 +910,21 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); - TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + final TestActorRef leaderCollector = + newLeaderCollectorActor(leaderActor.underlyingActor()); - TestActorRef followerCollector = actorFactory.createTestActor(MessageCollectorActor.props(). - withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); + final TestActorRef followerCollector = + actorFactory.createTestActor(MessageCollectorActor.props() + .withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); actorFactory.createTestActor( CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), - configParams, NO_PERSISTENCE, followerCollector).withDispatcher(Dispatchers.DefaultDispatcherId()), + configParams, NO_PERSISTENCE, followerCollector) + .withDispatcher(Dispatchers.DefaultDispatcherId()), followerActorId); leaderActor.tell(new RemoveServer(LEADER_ID), testKit.getRef()); - RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), + RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, removeServerReply.getStatus()); final ApplyState applyState = MessageCollectorActor.expectFirstMatching(followerCollector, ApplyState.class); @@ -940,7 +947,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); leaderActor.tell(new RemoveServer(LEADER_ID), testKit.getRef()); - RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), + RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.NOT_SUPPORTED, removeServerReply.getStatus()); LOG.info("testRemoveServerLeaderWithNoFollowers ending"); @@ -961,25 +969,25 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, follower1ActorPath, - FOLLOWER_ID2, follower2ActorPath), new MockRaftActorContext()). - withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); + FOLLOWER_ID2, follower2ActorPath), new MockRaftActorContext()) + .withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); TestActorRef follower1Collector = actorFactory.createTestActor( MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); - TestActorRef follower1RaftActor = actorFactory.createTestActor( + final TestActorRef follower1RaftActor = actorFactory.createTestActor( CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), - FOLLOWER_ID2, follower2ActorPath), configParams, NO_PERSISTENCE, follower1Collector). - withDispatcher(Dispatchers.DefaultDispatcherId()), follower1ActorId); + FOLLOWER_ID2, follower2ActorPath), configParams, NO_PERSISTENCE, follower1Collector) + .withDispatcher(Dispatchers.DefaultDispatcherId()), follower1ActorId); TestActorRef follower2Collector = actorFactory.createTestActor( MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); - TestActorRef follower2RaftActor = actorFactory.createTestActor( + final TestActorRef follower2RaftActor = actorFactory.createTestActor( CollectingMockRaftActor.props(FOLLOWER_ID2, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), - FOLLOWER_ID, follower1ActorPath), configParams, NO_PERSISTENCE, follower2Collector). - withDispatcher(Dispatchers.DefaultDispatcherId()), follower2ActorId); + FOLLOWER_ID, follower1ActorPath), configParams, NO_PERSISTENCE, follower2Collector) + .withDispatcher(Dispatchers.DefaultDispatcherId()), follower2ActorId); // Send first ChangeServersVotingStatus message @@ -994,12 +1002,14 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { votingServer(LEADER_ID), nonVotingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); MessageCollectorActor.expectFirstMatching(follower1Collector, ApplyState.class); - verifyServerConfigurationPayloadEntry(follower1RaftActor.underlyingActor().getRaftActorContext().getReplicatedLog(), - votingServer(LEADER_ID), nonVotingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); + verifyServerConfigurationPayloadEntry(follower1RaftActor.underlyingActor().getRaftActorContext() + .getReplicatedLog(), votingServer(LEADER_ID), nonVotingServer(FOLLOWER_ID), + nonVotingServer(FOLLOWER_ID2)); MessageCollectorActor.expectFirstMatching(follower2Collector, ApplyState.class); - verifyServerConfigurationPayloadEntry(follower2RaftActor.underlyingActor().getRaftActorContext().getReplicatedLog(), - votingServer(LEADER_ID), nonVotingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); + verifyServerConfigurationPayloadEntry(follower2RaftActor.underlyingActor().getRaftActorContext() + .getReplicatedLog(), votingServer(LEADER_ID), nonVotingServer(FOLLOWER_ID), + nonVotingServer(FOLLOWER_ID2)); MessageCollectorActor.clearMessages(leaderCollector); MessageCollectorActor.clearMessages(follower1Collector); @@ -1016,12 +1026,12 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { votingServer(LEADER_ID), votingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); MessageCollectorActor.expectFirstMatching(follower1Collector, ApplyState.class); - verifyServerConfigurationPayloadEntry(follower1RaftActor.underlyingActor().getRaftActorContext().getReplicatedLog(), - votingServer(LEADER_ID), votingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); + verifyServerConfigurationPayloadEntry(follower1RaftActor.underlyingActor().getRaftActorContext() + .getReplicatedLog(), votingServer(LEADER_ID), votingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); MessageCollectorActor.expectFirstMatching(follower2Collector, ApplyState.class); - verifyServerConfigurationPayloadEntry(follower2RaftActor.underlyingActor().getRaftActorContext().getReplicatedLog(), - votingServer(LEADER_ID), votingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); + verifyServerConfigurationPayloadEntry(follower2RaftActor.underlyingActor().getRaftActorContext() + .getReplicatedLog(), votingServer(LEADER_ID), votingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); LOG.info("testChangeServersVotingStatus ending"); } @@ -1040,25 +1050,25 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, follower1ActorPath, - FOLLOWER_ID2, follower2ActorPath), new MockRaftActorContext()). - withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); + FOLLOWER_ID2, follower2ActorPath), new MockRaftActorContext()) + .withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); TestActorRef follower1Collector = actorFactory.createTestActor( MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); - TestActorRef follower1RaftActor = actorFactory.createTestActor( + final TestActorRef follower1RaftActor = actorFactory.createTestActor( CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), - FOLLOWER_ID2, follower2ActorPath), configParams, NO_PERSISTENCE, follower1Collector). - withDispatcher(Dispatchers.DefaultDispatcherId()), follower1ActorId); + FOLLOWER_ID2, follower2ActorPath), configParams, NO_PERSISTENCE, follower1Collector) + .withDispatcher(Dispatchers.DefaultDispatcherId()), follower1ActorId); TestActorRef follower2Collector = actorFactory.createTestActor( MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); - TestActorRef follower2RaftActor = actorFactory.createTestActor( + final TestActorRef follower2RaftActor = actorFactory.createTestActor( CollectingMockRaftActor.props(FOLLOWER_ID2, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), - FOLLOWER_ID, follower1ActorPath), configParams, NO_PERSISTENCE, follower2Collector). - withDispatcher(Dispatchers.DefaultDispatcherId()), follower2ActorId); + FOLLOWER_ID, follower1ActorPath), configParams, NO_PERSISTENCE, follower2Collector) + .withDispatcher(Dispatchers.DefaultDispatcherId()), follower2ActorId); // Send ChangeServersVotingStatus message @@ -1071,12 +1081,12 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { nonVotingServer(LEADER_ID), votingServer(FOLLOWER_ID), votingServer(FOLLOWER_ID2)); MessageCollectorActor.expectFirstMatching(follower1Collector, ApplyState.class); - verifyServerConfigurationPayloadEntry(follower1RaftActor.underlyingActor().getRaftActorContext().getReplicatedLog(), - nonVotingServer(LEADER_ID), votingServer(FOLLOWER_ID), votingServer(FOLLOWER_ID2)); + verifyServerConfigurationPayloadEntry(follower1RaftActor.underlyingActor().getRaftActorContext() + .getReplicatedLog(), nonVotingServer(LEADER_ID), votingServer(FOLLOWER_ID), votingServer(FOLLOWER_ID2)); MessageCollectorActor.expectFirstMatching(follower2Collector, ApplyState.class); - verifyServerConfigurationPayloadEntry(follower2RaftActor.underlyingActor().getRaftActorContext().getReplicatedLog(), - nonVotingServer(LEADER_ID), votingServer(FOLLOWER_ID), votingServer(FOLLOWER_ID2)); + verifyServerConfigurationPayloadEntry(follower2RaftActor.underlyingActor().getRaftActorContext() + .getReplicatedLog(), nonVotingServer(LEADER_ID), votingServer(FOLLOWER_ID), votingServer(FOLLOWER_ID2)); verifyRaftState(RaftState.Leader, follower1RaftActor.underlyingActor(), follower2RaftActor.underlyingActor()); verifyRaftState(RaftState.Follower, leaderActor.underlyingActor()); @@ -1091,8 +1101,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { LOG.info("testChangeLeaderToNonVotingInSingleNode starting"); TestActorRef leaderActor = actorFactory.createTestActor( - MockLeaderRaftActor.props(ImmutableMap.of(), new MockRaftActorContext()). - withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); + MockLeaderRaftActor.props(ImmutableMap.of(), new MockRaftActorContext()) + .withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); leaderActor.tell(new ChangeServersVotingStatus(ImmutableMap.of(LEADER_ID, false)), testKit.getRef()); ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); @@ -1119,7 +1129,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList( new ServerInfo(node1ID, false), new ServerInfo(node2ID, false), new ServerInfo("downNode1", true), new ServerInfo("downNode2", true))); - ReplicatedLogImplEntry persistedServerConfigEntry = new ReplicatedLogImplEntry(0, 1, persistedServerConfig); + SimpleReplicatedLogEntry persistedServerConfigEntry = new SimpleReplicatedLogEntry(0, 1, persistedServerConfig); InMemoryJournal.addEntry(node1ID, 1, new UpdateElectionTerm(1, "downNode1")); InMemoryJournal.addEntry(node1ID, 2, persistedServerConfigEntry); @@ -1194,7 +1204,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); - ApplyJournalEntries apply = MessageCollectorActor.expectFirstMatching(node1Collector, ApplyJournalEntries.class); + ApplyJournalEntries apply = MessageCollectorActor.expectFirstMatching(node1Collector, + ApplyJournalEntries.class); assertEquals("getToIndex", 1, apply.getToIndex()); verifyServerConfigurationPayloadEntry(node1RaftActor.getRaftActorContext().getReplicatedLog(), votingServer(node1ID), votingServer(node2ID), nonVotingServer("downNode1"), @@ -1220,12 +1231,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { final String node1ID = "node1"; final String node2ID = "node2"; - PeerAddressResolver peerAddressResolver = peerId -> peerId.equals(node1ID) ? actorFactory.createTestActorPath(node1ID) : - peerId.equals(node2ID) ? actorFactory.createTestActorPath(node2ID) : null; + final PeerAddressResolver peerAddressResolver = peerId -> peerId.equals(node1ID) + ? actorFactory.createTestActorPath(node1ID) : peerId.equals(node2ID) + ? actorFactory.createTestActorPath(node2ID) : null; ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList( new ServerInfo(node1ID, false), new ServerInfo(node2ID, true))); - ReplicatedLogImplEntry persistedServerConfigEntry = new ReplicatedLogImplEntry(0, 1, persistedServerConfig); + SimpleReplicatedLogEntry persistedServerConfigEntry = new SimpleReplicatedLogEntry(0, 1, persistedServerConfig); InMemoryJournal.addEntry(node1ID, 1, new UpdateElectionTerm(1, "node1")); InMemoryJournal.addEntry(node1ID, 2, persistedServerConfigEntry); @@ -1242,7 +1254,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef node1RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node1ID, ImmutableMap.of(), configParams1, PERSISTENT, node1Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node1ID); - CollectingMockRaftActor node1RaftActor = node1RaftActorRef.underlyingActor(); + final CollectingMockRaftActor node1RaftActor = node1RaftActorRef.underlyingActor(); DefaultConfigParamsImpl configParams2 = new DefaultConfigParamsImpl(); configParams2.setElectionTimeoutFactor(1000000); @@ -1282,8 +1294,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { final String node1ID = "node1"; final String node2ID = "node2"; - PeerAddressResolver peerAddressResolver = peerId -> peerId.equals(node1ID) ? actorFactory.createTestActorPath(node1ID) : - peerId.equals(node2ID) ? actorFactory.createTestActorPath(node2ID) : null; + final PeerAddressResolver peerAddressResolver = peerId -> peerId.equals(node1ID) + ? actorFactory.createTestActorPath(node1ID) : peerId.equals(node2ID) + ? actorFactory.createTestActorPath(node2ID) : null; DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); @@ -1292,13 +1305,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList( new ServerInfo(node1ID, false), new ServerInfo(node2ID, false))); - ReplicatedLogImplEntry persistedServerConfigEntry = new ReplicatedLogImplEntry(0, 1, persistedServerConfig); + SimpleReplicatedLogEntry persistedServerConfigEntry = new SimpleReplicatedLogEntry(0, 1, persistedServerConfig); InMemoryJournal.addEntry(node1ID, 1, new UpdateElectionTerm(1, "node1")); InMemoryJournal.addEntry(node1ID, 2, persistedServerConfigEntry); InMemoryJournal.addEntry(node2ID, 1, new UpdateElectionTerm(1, "node1")); InMemoryJournal.addEntry(node2ID, 2, persistedServerConfigEntry); - InMemoryJournal.addEntry(node2ID, 3, new ReplicatedLogImplEntry(1, 1, + InMemoryJournal.addEntry(node2ID, 3, new SimpleReplicatedLogEntry(1, 1, new MockRaftActorContext.MockPayload("2"))); InMemoryJournal.addEntry(node2ID, 4, new ApplyJournalEntries(1)); @@ -1308,7 +1321,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef node1RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node1ID, ImmutableMap.of(), configParams, PERSISTENT, node1Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node1ID); - CollectingMockRaftActor node1RaftActor = node1RaftActorRef.underlyingActor(); + final CollectingMockRaftActor node1RaftActor = node1RaftActorRef.underlyingActor(); TestActorRef node2Collector = actorFactory.createTestActor( MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), @@ -1316,7 +1329,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef node2RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node2ID, ImmutableMap.of(), configParams, PERSISTENT, node2Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node2ID); - CollectingMockRaftActor node2RaftActor = node2RaftActorRef.underlyingActor(); + final CollectingMockRaftActor node2RaftActor = node2RaftActorRef.underlyingActor(); // Send a ChangeServersVotingStatus message to node1 to change mode1 to voting. This should cause // node1 to try to elect itself as leader in order to apply the new server config. However node1's log @@ -1354,12 +1367,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { final String node1ID = "node1"; final String node2ID = "node2"; - configParams.setPeerAddressResolver(peerId -> peerId.equals(node1ID) ? actorFactory.createTestActorPath(node1ID) : - peerId.equals(node2ID) ? actorFactory.createTestActorPath(node2ID) : null); + configParams.setPeerAddressResolver(peerId -> peerId.equals(node1ID) + ? actorFactory.createTestActorPath(node1ID) : peerId.equals(node2ID) + ? actorFactory.createTestActorPath(node2ID) : null); ServerConfigurationPayload persistedServerConfig = new ServerConfigurationPayload(Arrays.asList( new ServerInfo(node1ID, false), new ServerInfo(node2ID, true))); - ReplicatedLogImplEntry persistedServerConfigEntry = new ReplicatedLogImplEntry(0, 1, persistedServerConfig); + SimpleReplicatedLogEntry persistedServerConfigEntry = new SimpleReplicatedLogEntry(0, 1, persistedServerConfig); InMemoryJournal.addEntry(node1ID, 1, new UpdateElectionTerm(1, "node1")); InMemoryJournal.addEntry(node1ID, 2, persistedServerConfigEntry); @@ -1372,7 +1386,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { TestActorRef node1RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node1ID, ImmutableMap.of(), configParams, PERSISTENT, node1Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node1ID); - CollectingMockRaftActor node1RaftActor = node1RaftActorRef.underlyingActor(); + final CollectingMockRaftActor node1RaftActor = node1RaftActorRef.underlyingActor(); TestActorRef node2Collector = actorFactory.createTestActor( MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), @@ -1416,9 +1430,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { private static void verifyRaftState(RaftState expState, RaftActor... raftActors) { Stopwatch sw = Stopwatch.createStarted(); - while(sw.elapsed(TimeUnit.SECONDS) <= 5) { - for(RaftActor raftActor: raftActors) { - if(raftActor.getRaftState() == expState) { + while (sw.elapsed(TimeUnit.SECONDS) <= 5) { + for (RaftActor raftActor : raftActors) { + if (raftActor.getRaftState() == expState) { return; } } @@ -1462,17 +1476,18 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ElectionTermImpl termInfo = new ElectionTermImpl(noPersistence, id, LOG); termInfo.update(1, LEADER_ID); return new RaftActorContextImpl(actor, actor.underlyingActor().getContext(), - id, termInfo, -1, -1, ImmutableMap.of(LEADER_ID, ""), configParams, noPersistence, LOG); + id, termInfo, -1, -1, ImmutableMap.of(LEADER_ID, ""), configParams, + noPersistence, applyState -> actor.tell(applyState, actor), LOG); } - static abstract class AbstractMockRaftActor extends MockRaftActor { + abstract static class AbstractMockRaftActor extends MockRaftActor { private volatile TestActorRef collectorActor; private volatile Class dropMessageOfType; AbstractMockRaftActor(String id, Map peerAddresses, Optional config, boolean persistent, TestActorRef collectorActor) { - super(builder().id(id).peerAddresses(peerAddresses).config(config.get()). - persistent(Optional.of(persistent))); + super(builder().id(id).peerAddresses(peerAddresses).config(config.get()) + .persistent(Optional.of(persistent))); this.collectorActor = collectorActor; } @@ -1486,11 +1501,11 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Override public void handleCommand(Object message) { - if(dropMessageOfType == null || !dropMessageOfType.equals(message.getClass())) { + if (dropMessageOfType == null || !dropMessageOfType.equals(message.getClass())) { super.handleCommand(message); } - if(collectorActor != null) { + if (collectorActor != null) { collectorActor.tell(message, getSender()); } } @@ -1514,7 +1529,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } public static Props props(final String id, final Map peerAddresses, - ConfigParams config, boolean persistent, TestActorRef collectorActor){ + ConfigParams config, boolean persistent, TestActorRef collectorActor) { return Props.create(CollectingMockRaftActor.class, id, peerAddresses, Optional.of(config), persistent, collectorActor); @@ -1529,7 +1544,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { setPersistence(false); RaftActorContext context = getRaftActorContext(); - for(int i = 0; i < fromContext.getReplicatedLog().size(); i++) { + for (int i = 0; i < fromContext.getReplicatedLog().size(); i++) { ReplicatedLogEntry entry = fromContext.getReplicatedLog().get(i); getState().add(entry.getData()); context.getReplicatedLog().append(entry); @@ -1548,6 +1563,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Override + @SuppressWarnings("checkstyle:IllegalCatch") public void createSnapshot(ActorRef actorRef) { try { actorRef.tell(new CaptureSnapshotReply(RaftActorTest.fromObject(getState()).toByteArray()), actorRef); @@ -1566,7 +1582,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { public static class MockNewFollowerRaftActor extends AbstractMockRaftActor { public MockNewFollowerRaftActor(ConfigParams config, TestActorRef collectorActor) { - super(NEW_SERVER_ID, Maps.newHashMap(), Optional.of(config), NO_PERSISTENCE, collectorActor); + super(NEW_SERVER_ID, Maps.newHashMap(), Optional.of(config), NO_PERSISTENCE, + collectorActor); setPersistence(false); }