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=54ba69ec84025a05496f9b761dc8da407ded1e60;hp=02ba0153191c7d8fec0dad94ca02729c4b46f24c;hb=refs%2Fchanges%2F09%2F83009%2F6;hpb=782f06145be869e3f541a199915212f8188488ae 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 02ba015319..54ba69ec84 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,35 +14,38 @@ 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.AbstractActor; import akka.actor.ActorRef; import akka.actor.Props; -import akka.actor.UntypedActor; import akka.dispatch.Dispatchers; -import akka.testkit.JavaTestKit; import akka.testkit.TestActorRef; -import com.google.common.base.Optional; +import akka.testkit.javadsl.TestKit; import com.google.common.base.Stopwatch; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; import com.google.common.collect.Sets; +import com.google.common.io.ByteSource; +import com.google.common.util.concurrent.MoreExecutors; +import java.io.OutputStream; +import java.time.Duration; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.TimeUnit; +import org.apache.commons.lang3.SerializationUtils; import org.junit.After; import org.junit.Before; import org.junit.Test; import org.opendaylight.controller.cluster.NonPersistentDataProvider; -import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload.ServerInfo; -import org.opendaylight.controller.cluster.raft.base.messages.ApplyJournalEntries; import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot; import org.opendaylight.controller.cluster.raft.base.messages.ApplyState; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; -import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout; import org.opendaylight.controller.cluster.raft.base.messages.InitiateCaptureSnapshot; -import org.opendaylight.controller.cluster.raft.base.messages.SnapshotComplete; -import org.opendaylight.controller.cluster.raft.base.messages.UpdateElectionTerm; +import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow; import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader; import org.opendaylight.controller.cluster.raft.behaviors.Follower; import org.opendaylight.controller.cluster.raft.behaviors.Leader; @@ -59,6 +62,12 @@ import org.opendaylight.controller.cluster.raft.messages.ServerChangeReply; import org.opendaylight.controller.cluster.raft.messages.ServerChangeStatus; import org.opendaylight.controller.cluster.raft.messages.ServerRemoved; import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSnapshotReply; +import org.opendaylight.controller.cluster.raft.persisted.ApplyJournalEntries; +import org.opendaylight.controller.cluster.raft.persisted.ByteState; +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; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; @@ -91,10 +100,10 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(FOLLOWER_ID)); private TestActorRef newFollowerRaftActor; - private TestActorRef newFollowerCollectorActor; + private ActorRef newFollowerCollectorActor; private RaftActorContext newFollowerActorContext; - private final JavaTestKit testKit = new JavaTestKit(getSystem()); + private final TestKit testKit = new TestKit(getSystem()); @Before public void setup() { @@ -102,11 +111,11 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { InMemorySnapshotStore.clear(); } + @SuppressWarnings("checkstyle:IllegalCatch") private void setupNewFollower() { DefaultConfigParamsImpl configParams = newFollowerConfigParams(); - newFollowerCollectorActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + newFollowerCollectorActor = actorFactory.createActor(MessageCollectorActor.props(), actorFactory.generateActorId(NEW_SERVER_ID + "Collector")); newFollowerRaftActor = actorFactory.createTestActor(MockNewFollowerRaftActor.props( configParams, newFollowerCollectorActor).withDispatcher(Dispatchers.DefaultDispatcherId()), @@ -128,12 +137,12 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @After - public void tearDown() throws Exception { + public void tearDown() { actorFactory.close(); } @Test - public void testAddServerWithExistingFollower() throws Exception { + public void testAddServerWithExistingFollower() { LOG.info("testAddServerWithExistingFollower starting"); setupNewFollower(); RaftActorContextImpl followerActorContext = newFollowerContext(FOLLOWER_ID, followerActor); @@ -156,18 +165,17 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { clearMessages(followerActor); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final ActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); // Leader should install snapshot - capture and verify ApplySnapshot contents ApplySnapshot applySnapshot = expectFirstMatching(newFollowerCollectorActor, ApplySnapshot.class); - @SuppressWarnings("unchecked") - List snapshotState = (List) MockRaftActor.toObject(applySnapshot.getSnapshot().getState()); + List snapshotState = MockRaftActor.fromState(applySnapshot.getSnapshot().getState()); assertEquals("Snapshot state", snapshotState, leaderRaftActor.getState()); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); @@ -197,36 +205,29 @@ 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"); } @Test - public void testAddServerWithNoExistingFollower() throws Exception { + public void testAddServerWithNoExistingFollower() { LOG.info("testAddServerWithNoExistingFollower starting"); setupNewFollower(); @@ -242,20 +243,19 @@ 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 ActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); // Leader should install snapshot - capture and verify ApplySnapshot contents ApplySnapshot applySnapshot = expectFirstMatching(newFollowerCollectorActor, ApplySnapshot.class); - @SuppressWarnings("unchecked") - List snapshotState = (List) MockRaftActor.toObject(applySnapshot.getSnapshot().getState()); + List snapshotState = MockRaftActor.fromState(applySnapshot.getSnapshot().getState()); assertEquals("Snapshot state", snapshotState, leaderRaftActor.getState()); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); @@ -283,7 +283,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServersAsNonVoting() throws Exception { + public void testAddServersAsNonVoting() { LOG.info("testAddServersAsNonVoting starting"); setupNewFollower(); @@ -295,13 +295,13 @@ 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 ActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), false), testKit.getRef()); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); @@ -338,9 +338,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderActor.tell(new AddServer(NEW_SERVER_ID2, followerActor.path().toString(), false), testKit.getRef()); - addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); - assertEquals("getLeaderHint", java.util.Optional.of(LEADER_ID), addServerReply.getLeaderHint()); + assertEquals("getLeaderHint", Optional.of(LEADER_ID), addServerReply.getLeaderHint()); expectFirstMatching(leaderCollectorActor, ApplyState.class); assertEquals("Leader journal last index", 1, leaderActorContext.getReplicatedLog().lastIndex()); @@ -353,7 +353,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithOperationInProgress() throws Exception { + public void testAddServerWithOperationInProgress() { LOG.info("testAddServerWithOperationInProgress starting"); setupNewFollower(); @@ -365,16 +365,16 @@ 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 ActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); RaftActorContext follower2ActorContext = newFollowerContext(NEW_SERVER_ID2, followerActor); Follower newFollower2 = new Follower(follower2ActorContext); followerActor.underlyingActor().setBehavior(newFollower2); MockNewFollowerRaftActor newFollowerRaftActorInstance = newFollowerRaftActor.underlyingActor(); - newFollowerRaftActorInstance.setDropMessageOfType(InstallSnapshot.SERIALIZABLE_CLASS); + newFollowerRaftActorInstance.setDropMessageOfType(InstallSnapshot.class); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); @@ -383,7 +383,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { InstallSnapshot installSnapshot = expectFirstMatching(newFollowerCollectorActor, InstallSnapshot.class); // Send a second AddServer - should get queued - JavaTestKit testKit2 = new JavaTestKit(getSystem()); + TestKit testKit2 = new TestKit(getSystem()); leaderActor.tell(new AddServer(NEW_SERVER_ID2, followerActor.path().toString(), false), testKit2.getRef()); // Continue the first AddServer @@ -391,10 +391,10 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { newFollowerRaftActor.tell(installSnapshot, leaderActor); // Verify both complete successfully - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); - addServerReply = testKit2.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + addServerReply = testKit2.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); // Verify ServerConfigurationPayload entries in leader's log @@ -416,7 +416,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithPriorSnapshotInProgress() throws Exception { + public void testAddServerWithPriorSnapshotInProgress() { LOG.info("testAddServerWithPriorSnapshotInProgress starting"); setupNewFollower(); @@ -428,9 +428,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); + ActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop commit message for now to delay snapshot completion leaderRaftActor.setDropMessageOfType(String.class); @@ -444,7 +444,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderRaftActor.setDropMessageOfType(null); leaderActor.tell(commitMsg, leaderActor); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); @@ -463,7 +463,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithPriorSnapshotCompleteTimeout() throws Exception { + public void testAddServerWithPriorSnapshotCompleteTimeout() { LOG.info("testAddServerWithPriorSnapshotCompleteTimeout starting"); setupNewFollower(); @@ -486,7 +486,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.TIMEOUT, addServerReply.getStatus()); assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); @@ -495,7 +495,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithLeaderChangeBeforePriorSnapshotComplete() throws Exception { + public void testAddServerWithLeaderChangeBeforePriorSnapshotComplete() { LOG.info("testAddServerWithLeaderChangeBeforePriorSnapshotComplete starting"); setupNewFollower(); @@ -510,7 +510,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(100); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + final ActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop the commit message so the snapshot doesn't complete yet. leaderRaftActor.setDropMessageOfType(COMMIT_MESSAGE_CLASS); @@ -534,7 +534,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderActor.tell(new RaftActorServerConfigurationSupport.ServerOperationTimeout(NEW_SERVER_ID), leaderActor); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); @@ -544,7 +544,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithLeaderChangeDuringInstallSnapshot() throws Exception { + public void testAddServerWithLeaderChangeDuringInstallSnapshot() { LOG.info("testAddServerWithLeaderChangeDuringInstallSnapshot starting"); setupNewFollower(); @@ -560,14 +560,14 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(8); - TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + ActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop the UnInitializedFollowerSnapshotReply to delay it. leaderRaftActor.setDropMessageOfType(UnInitializedFollowerSnapshotReply.class); 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 @@ -580,7 +580,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderRaftActor.setDropMessageOfType(null); leaderActor.tell(snapshotReply, leaderActor); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); @@ -589,7 +589,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithInstallSnapshotTimeout() throws Exception { + public void testAddServerWithInstallSnapshotTimeout() { LOG.info("testAddServerWithInstallSnapshotTimeout starting"); setupNewFollower(); @@ -605,13 +605,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(1); // Drop the InstallSnapshot message so it times out - newFollowerRaftActor.underlyingActor().setDropMessageOfType(InstallSnapshot.SERIALIZABLE_CLASS); + newFollowerRaftActor.underlyingActor().setDropMessageOfType(InstallSnapshot.class); leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); leaderActor.tell(new UnInitializedFollowerSnapshotReply("bogus"), leaderActor); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.TIMEOUT, addServerReply.getStatus()); assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); @@ -631,13 +631,14 @@ 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()); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + noLeaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), + testKit.getRef()); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); LOG.info("testAddServerWithNoLeader ending"); @@ -656,16 +657,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 ActorRef 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); @@ -685,7 +685,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderActor.tell(new AddServer(NEW_SERVER_ID2, "", false), testKit.getRef()); // The first AddServer should succeed with OK even though consensus wasn't reached - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); @@ -694,12 +694,12 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { votingServer(NEW_SERVER_ID)); // The second AddServer should fail since consensus wasn't reached for the first - addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.PRIOR_REQUEST_CONSENSUS_TIMEOUT, addServerReply.getStatus()); // Re-send the second AddServer - should also fail leaderActor.tell(new AddServer(NEW_SERVER_ID2, "", false), testKit.getRef()); - addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.PRIOR_REQUEST_CONSENSUS_TIMEOUT, addServerReply.getStatus()); LOG.info("testAddServerWithNoConsensusReached ending"); @@ -718,7 +718,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { leaderActor.tell(new AddServer(FOLLOWER_ID, followerActor.path().toString(), true), testKit.getRef()); - AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + AddServerReply addServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.ALREADY_EXISTS, addServerReply.getStatus()); LOG.info("testAddServerWithExistingServer ending"); @@ -732,21 +732,21 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - TestActorRef leaderActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(LEADER_ID)); + ActorRef leaderActor = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId(LEADER_ID)); 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"); @@ -760,18 +760,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); @@ -788,13 +789,13 @@ 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(Duration.ofSeconds(5), RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, removeServerReply.getStatus()); LOG.info("testRemoveServerWithNoLeader ending"); @@ -812,7 +813,7 @@ 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(Duration.ofSeconds(5), RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.DOES_NOT_EXIST, removeServerReply.getStatus()); LOG.info("testRemoveServerNonExistentServer ending"); @@ -825,14 +826,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - TestActorRef leaderActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(LEADER_ID)); + ActorRef leaderActor = actorFactory.createTestActor( + MessageCollectorActor.props(), actorFactory.generateActorId(LEADER_ID)); 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(); @@ -853,38 +853,57 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); configParams.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName()); - final String followerActorId = actorFactory.generateActorId(FOLLOWER_ID); - final String followerActorPath = actorFactory.createTestActorPath(followerActorId); + final String follower1ActorId = actorFactory.generateActorId(FOLLOWER_ID); + final String follower1ActorPath = actorFactory.createTestActorPath(follower1ActorId); + final String follower2ActorId = actorFactory.generateActorId(FOLLOWER_ID2); + final String follower2ActorPath = actorFactory.createTestActorPath(follower2ActorId); RaftActorContext initialActorContext = new MockRaftActorContext(); - TestActorRef leaderActor = actorFactory.createTestActor( - MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActorPath), + final String downNodeId = "downNode"; + TestActorRef leaderActor = actorFactory.createTestActor(MockLeaderRaftActor.props( + ImmutableMap.of(FOLLOWER_ID, follower1ActorPath, FOLLOWER_ID2, follower2ActorPath, downNodeId, ""), initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); - TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + final ActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); - TestActorRef collector = - actorFactory.createTestActor(MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); - TestActorRef followerRaftActor = actorFactory.createTestActor( - CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), - configParams, NO_PERSISTENCE, collector).withDispatcher(Dispatchers.DefaultDispatcherId()), - followerActorId); + ActorRef follower1Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); + final TestActorRef follower1Actor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), + FOLLOWER_ID2, follower2ActorPath, downNodeId, ""), configParams, NO_PERSISTENCE, + follower1Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), follower1ActorId); + + ActorRef follower2Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); + final TestActorRef follower2Actor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID2, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), + FOLLOWER_ID, follower1ActorPath, downNodeId, ""), configParams, NO_PERSISTENCE, + follower2Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), follower2ActorId); + + leaderActor.underlyingActor().waitForInitializeBehaviorComplete(); + follower1Actor.underlyingActor().waitForInitializeBehaviorComplete(); + follower2Actor.underlyingActor().waitForInitializeBehaviorComplete(); leaderActor.tell(new RemoveServer(FOLLOWER_ID), testKit.getRef()); - RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + RemoveServerReply removeServerReply = testKit.expectMsgClass(Duration.ofSeconds(5), RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, removeServerReply.getStatus()); - final ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); + ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); + assertEquals(0L, applyState.getReplicatedLogEntry().getIndex()); + verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + votingServer(LEADER_ID), votingServer(FOLLOWER_ID2), votingServer(downNodeId)); + + applyState = MessageCollectorActor.expectFirstMatching(follower2Collector, ApplyState.class); assertEquals(0L, applyState.getReplicatedLogEntry().getIndex()); - verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), votingServer(LEADER_ID)); + verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + votingServer(LEADER_ID), votingServer(FOLLOWER_ID2), votingServer(downNodeId)); RaftActorBehavior currentBehavior = leaderActor.underlyingActor().getCurrentBehavior(); assertTrue("Expected Leader", currentBehavior instanceof Leader); - assertEquals("Follower ids size", 0, ((Leader)currentBehavior).getFollowerIds().size()); + assertEquals("Follower ids size", 2, ((Leader)currentBehavior).getFollowerIds().size()); - MessageCollectorActor.expectFirstMatching(collector, ServerRemoved.class); + MessageCollectorActor.expectFirstMatching(follower1Collector, ServerRemoved.class); LOG.info("testRemoveServer ending"); } @@ -906,17 +925,18 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); - TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + final ActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); - TestActorRef followerCollector = actorFactory.createTestActor(MessageCollectorActor.props(). - withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); + final ActorRef followerCollector = + actorFactory.createActor(MessageCollectorActor.props(), 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(Duration.ofSeconds(5), RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, removeServerReply.getStatus()); final ApplyState applyState = MessageCollectorActor.expectFirstMatching(followerCollector, ApplyState.class); @@ -939,7 +959,7 @@ 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(Duration.ofSeconds(5), RemoveServerReply.class); assertEquals("getStatus", ServerChangeStatus.NOT_SUPPORTED, removeServerReply.getStatus()); LOG.info("testRemoveServerLeaderWithNoFollowers ending"); @@ -960,31 +980,29 @@ 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)); - TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); - - TestActorRef follower1Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); - TestActorRef follower1RaftActor = actorFactory.createTestActor( + FOLLOWER_ID2, follower2ActorPath), new MockRaftActorContext()) + .withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); + ActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + + ActorRef follower1Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); + 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( + ActorRef follower2Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); + 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 leaderActor.tell(new ChangeServersVotingStatus(ImmutableMap.of(FOLLOWER_ID, false, FOLLOWER_ID2, false)), testKit.getRef()); - ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + ServerChangeReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); final ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); @@ -993,12 +1011,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); @@ -1007,7 +1027,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // Send second ChangeServersVotingStatus message leaderActor.tell(new ChangeServersVotingStatus(ImmutableMap.of(FOLLOWER_ID, true)), testKit.getRef()); - reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + reply = testKit.expectMsgClass(Duration.ofSeconds(5), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); @@ -1015,12 +1035,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"); } @@ -1039,30 +1059,28 @@ 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)); - TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); - - TestActorRef follower1Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); - TestActorRef follower1RaftActor = actorFactory.createTestActor( + FOLLOWER_ID2, follower2ActorPath), new MockRaftActorContext()) + .withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); + ActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + + ActorRef follower1Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); + 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( + ActorRef follower2Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); + 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 leaderActor.tell(new ChangeServersVotingStatus(ImmutableMap.of(LEADER_ID, false)), testKit.getRef()); - ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + ServerChangeReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); @@ -1070,12 +1088,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()); @@ -1085,12 +1103,28 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { LOG.info("testChangeLeaderToNonVoting ending"); } + @Test + public void testChangeLeaderToNonVotingInSingleNode() { + LOG.info("testChangeLeaderToNonVotingInSingleNode starting"); + + TestActorRef leaderActor = actorFactory.createTestActor( + 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(Duration.ofSeconds(5), ServerChangeReply.class); + assertEquals("getStatus", ServerChangeStatus.INVALID_REQUEST, reply.getStatus()); + + LOG.info("testChangeLeaderToNonVotingInSingleNode ending"); + } + @Test public void testChangeToVotingWithNoLeader() { LOG.info("testChangeToVotingWithNoLeader starting"); DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); + configParams.setElectionTimeoutFactor(5); final String node1ID = "node1"; final String node2ID = "node2"; @@ -1102,31 +1136,31 @@ 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); + InMemoryJournal.addEntry(node1ID, 3, new ApplyJournalEntries(0)); InMemoryJournal.addEntry(node2ID, 1, new UpdateElectionTerm(1, "downNode2")); InMemoryJournal.addEntry(node2ID, 2, persistedServerConfigEntry); + InMemoryJournal.addEntry(node2ID, 3, new ApplyJournalEntries(0)); - TestActorRef node1Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); + ActorRef node1Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); TestActorRef node1RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node1ID, ImmutableMap.of(), configParams, PERSISTENT, node1Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node1ID); CollectingMockRaftActor node1RaftActor = node1RaftActorRef.underlyingActor(); - TestActorRef node2Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); + ActorRef node2Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); TestActorRef node2RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node2ID, ImmutableMap.of(), configParams, PERSISTENT, node2Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node2ID); CollectingMockRaftActor node2RaftActor = node2RaftActorRef.underlyingActor(); - // Wait for snapshot after recovery - MessageCollectorActor.expectFirstMatching(node1Collector, SnapshotComplete.class); + node1RaftActor.waitForInitializeBehaviorComplete(); + node2RaftActor.waitForInitializeBehaviorComplete(); // Verify the intended server config was loaded and applied. verifyServerConfigurationPayloadEntry(node1RaftActor.getRaftActorContext().getReplicatedLog(), @@ -1136,7 +1170,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { assertEquals("getRaftState", RaftState.Follower, node1RaftActor.getRaftState()); assertEquals("getLeaderId", null, node1RaftActor.getLeaderId()); - MessageCollectorActor.expectFirstMatching(node2Collector, SnapshotComplete.class); + verifyServerConfigurationPayloadEntry(node2RaftActor.getRaftActorContext().getReplicatedLog(), + nonVotingServer(node1ID), nonVotingServer(node2ID), votingServer("downNode1"), + votingServer("downNode2")); assertEquals("isVotingMember", false, node2RaftActor.getRaftActorContext().isVotingMember()); // For the test, we send a ChangeServersVotingStatus message to node1 to flip the voting states for @@ -1149,18 +1185,32 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ChangeServersVotingStatus changeServers = new ChangeServersVotingStatus(ImmutableMap.of(node1ID, true, node2ID, true, "downNode1", false, "downNode2", false)); node1RaftActorRef.tell(changeServers, testKit.getRef()); - ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + ServerChangeReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, reply.getStatus()); + assertEquals("getRaftState", RaftState.Follower, node1RaftActor.getRaftState()); + + // Send an AppendEntries so node1 has a leaderId + + long term = node1RaftActor.getRaftActorContext().getTermInformation().getCurrentTerm(); + node1RaftActorRef.tell(new AppendEntries(term, "downNode1", -1L, -1L, + Collections.emptyList(), 0, -1, (short)1), ActorRef.noSender()); + + // Wait for the ElectionTimeout to clear the leaderId. The leaderId must be null so on the next + // ChangeServersVotingStatus message, it will try to elect a leader. + + AbstractRaftActorIntegrationTest.verifyRaftState(node1RaftActorRef, + rs -> assertEquals("getLeader", null, rs.getLeader())); // Update node2's peer address and send the message again node1RaftActor.setPeerAddress(node2ID, node2RaftActorRef.path().toString()); node1RaftActorRef.tell(changeServers, testKit.getRef()); - reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + reply = testKit.expectMsgClass(Duration.ofSeconds(5), 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"), @@ -1186,12 +1236,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); @@ -1202,28 +1253,23 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { configParams1.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); configParams1.setElectionTimeoutFactor(1); configParams1.setPeerAddressResolver(peerAddressResolver); - TestActorRef node1Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); + ActorRef node1Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); 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); configParams2.setPeerAddressResolver(peerAddressResolver); - TestActorRef node2Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); + ActorRef node2Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); TestActorRef node2RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node2ID, ImmutableMap.of(), configParams2, PERSISTENT, node2Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node2ID); CollectingMockRaftActor node2RaftActor = node2RaftActorRef.underlyingActor(); - // Wait for snapshot after recovery - MessageCollectorActor.expectFirstMatching(node1Collector, SnapshotComplete.class); - // 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. But we'll drop // RequestVote messages in node2 which should cause node1 to time out and revert back to the previous @@ -1234,7 +1280,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ChangeServersVotingStatus changeServers = new ChangeServersVotingStatus(ImmutableMap.of(node1ID, true)); node1RaftActorRef.tell(changeServers, testKit.getRef()); - ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + ServerChangeReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, reply.getStatus()); assertEquals("Server config", Sets.newHashSet(nonVotingServer(node1ID), votingServer(node2ID)), @@ -1251,8 +1297,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)); @@ -1261,33 +1308,29 @@ 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)); - TestActorRef node1Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); + ActorRef node1Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); 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()), - actorFactory.generateActorId("collector")); + ActorRef node2Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); TestActorRef node2RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node2ID, ImmutableMap.of(), configParams, PERSISTENT, node2Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node2ID); - CollectingMockRaftActor node2RaftActor = node2RaftActorRef.underlyingActor(); - - // Wait for snapshot after recovery - MessageCollectorActor.expectFirstMatching(node1Collector, SnapshotComplete.class); + 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 @@ -1297,7 +1340,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ChangeServersVotingStatus changeServers = new ChangeServersVotingStatus( ImmutableMap.of(node1ID, true, node2ID, true)); node1RaftActorRef.tell(changeServers, testKit.getRef()); - ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + ServerChangeReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); MessageCollectorActor.expectFirstMatching(node2Collector, ApplyJournalEntries.class); @@ -1325,37 +1368,33 @@ 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); InMemoryJournal.addEntry(node2ID, 1, new UpdateElectionTerm(1, "node1")); InMemoryJournal.addEntry(node2ID, 2, persistedServerConfigEntry); - TestActorRef node1Collector = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId("collector")); + ActorRef node1Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); 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()), - actorFactory.generateActorId("collector")); + ActorRef node2Collector = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("collector")); TestActorRef node2RaftActorRef = actorFactory.createTestActor( CollectingMockRaftActor.props(node2ID, ImmutableMap.of(), configParams, PERSISTENT, node2Collector).withDispatcher(Dispatchers.DefaultDispatcherId()), node2ID); CollectingMockRaftActor node2RaftActor = node2RaftActorRef.underlyingActor(); - // Wait for snapshot after recovery - MessageCollectorActor.expectFirstMatching(node1Collector, SnapshotComplete.class); - // Send a ChangeServersVotingStatus message to node1 to change node1 to voting. This should cause // node1 to try to elect itself as leader in order to apply the new server config. But we'll drop // RequestVote messages in node2 and make it the leader so node1 should forward the server change @@ -1369,9 +1408,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { MessageCollectorActor.expectFirstMatching(node2Collector, RequestVote.class); - node2RaftActorRef.tell(ElectionTimeout.INSTANCE, ActorRef.noSender()); + node2RaftActorRef.tell(TimeoutNow.INSTANCE, ActorRef.noSender()); - ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + ServerChangeReply reply = testKit.expectMsgClass(Duration.ofSeconds(5), ServerChangeReply.class); assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); MessageCollectorActor.expectFirstMatching(node1Collector, ApplyJournalEntries.class); @@ -1388,11 +1427,11 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { LOG.info("testChangeToVotingWithNoLeaderAndOtherLeaderElected ending"); } - private static void verifyRaftState(RaftState expState, RaftActor... raftActors) { + private static void verifyRaftState(final RaftState expState, final 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; } } @@ -1401,70 +1440,71 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { fail("None of the RaftActors have state " + expState); } - private static ServerInfo votingServer(String id) { + private static ServerInfo votingServer(final String id) { return new ServerInfo(id, true); } - private static ServerInfo nonVotingServer(String id) { + private static ServerInfo nonVotingServer(final String id) { return new ServerInfo(id, false); } - private TestActorRef newLeaderCollectorActor(MockLeaderRaftActor leaderRaftActor) { + private ActorRef newLeaderCollectorActor(final MockLeaderRaftActor leaderRaftActor) { return newCollectorActor(leaderRaftActor, LEADER_ID); } - private TestActorRef newCollectorActor(AbstractMockRaftActor raftActor, String id) { - TestActorRef collectorActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(id + "Collector")); + private ActorRef newCollectorActor(final AbstractMockRaftActor raftActor, final String id) { + ActorRef collectorActor = actorFactory.createTestActor( + MessageCollectorActor.props(), actorFactory.generateActorId(id + "Collector")); raftActor.setCollectorActor(collectorActor); return collectorActor; } - private static void verifyServerConfigurationPayloadEntry(ReplicatedLog log, ServerInfo... expected) { + private static void verifyServerConfigurationPayloadEntry(final ReplicatedLog log, final ServerInfo... expected) { ReplicatedLogEntry logEntry = log.get(log.lastIndex()); assertEquals("Last log entry payload class", ServerConfigurationPayload.class, logEntry.getData().getClass()); ServerConfigurationPayload payload = (ServerConfigurationPayload)logEntry.getData(); assertEquals("Server config", Sets.newHashSet(expected), Sets.newHashSet(payload.getServerConfig())); } - private static RaftActorContextImpl newFollowerContext(String id, TestActorRef actor) { + private static RaftActorContextImpl newFollowerContext(final String id, + final TestActorRef actor) { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); configParams.setElectionTimeoutFactor(100000); - NonPersistentDataProvider noPersistence = new NonPersistentDataProvider(); + NonPersistentDataProvider noPersistence = new NonPersistentDataProvider(Runnable::run); 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, MoreExecutors.directExecutor()); } - static abstract class AbstractMockRaftActor extends MockRaftActor { - private volatile TestActorRef collectorActor; + abstract static class AbstractMockRaftActor extends MockRaftActor { + private volatile ActorRef 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))); + AbstractMockRaftActor(final String id, final Map peerAddresses, + final Optional config, final boolean persistent, final ActorRef collectorActor) { + super(builder().id(id).peerAddresses(peerAddresses).config(config.get()) + .persistent(Optional.of(persistent))); this.collectorActor = collectorActor; } - void setDropMessageOfType(Class dropMessageOfType) { + void setDropMessageOfType(final Class dropMessageOfType) { this.dropMessageOfType = dropMessageOfType; } - void setCollectorActor(TestActorRef collectorActor) { + void setCollectorActor(final ActorRef collectorActor) { this.collectorActor = collectorActor; } @Override - public void handleCommand(Object message) { - if(dropMessageOfType == null || !dropMessageOfType.equals(message.getClass())) { + public void handleCommand(final Object message) { + if (dropMessageOfType == null || !dropMessageOfType.equals(message.getClass())) { super.handleCommand(message); } - if(collectorActor != null) { + if (collectorActor != null) { collectorActor.tell(message, getSender()); } } @@ -1472,23 +1512,31 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { public static class CollectingMockRaftActor extends AbstractMockRaftActor { - CollectingMockRaftActor(String id, Map peerAddresses, Optional config, - boolean persistent, TestActorRef collectorActor) { + CollectingMockRaftActor(final String id, final Map peerAddresses, + final Optional config, final boolean persistent, final ActorRef collectorActor) { super(id, peerAddresses, config, persistent, collectorActor); snapshotCohortDelegate = new RaftActorSnapshotCohort() { @Override - public void createSnapshot(ActorRef actorRef) { - actorRef.tell(new CaptureSnapshotReply(new byte[0]), actorRef); + public void createSnapshot(final ActorRef actorRef, + final Optional installSnapshotStream) { + actorRef.tell(new CaptureSnapshotReply(ByteState.empty(), installSnapshotStream), actorRef); } @Override - public void applySnapshot(byte[] snapshotBytes) { + public void applySnapshot( + final org.opendaylight.controller.cluster.raft.persisted.Snapshot.State snapshotState) { + } + + @Override + public org.opendaylight.controller.cluster.raft.persisted.Snapshot.State deserializeSnapshot( + final ByteSource snapshotBytes) { + throw new UnsupportedOperationException(); } }; } public static Props props(final String id, final Map peerAddresses, - ConfigParams config, boolean persistent, TestActorRef collectorActor){ + final ConfigParams config, final boolean persistent, final ActorRef collectorActor) { return Props.create(CollectingMockRaftActor.class, id, peerAddresses, Optional.of(config), persistent, collectorActor); @@ -1497,13 +1545,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } public static class MockLeaderRaftActor extends AbstractMockRaftActor { - public MockLeaderRaftActor(Map peerAddresses, ConfigParams config, - RaftActorContext fromContext) { + public MockLeaderRaftActor(final Map peerAddresses, final ConfigParams config, + final RaftActorContext fromContext) { super(LEADER_ID, peerAddresses, Optional.of(config), NO_PERSISTENCE, null); 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); @@ -1522,15 +1570,17 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Override - public void createSnapshot(ActorRef actorRef) { - try { - actorRef.tell(new CaptureSnapshotReply(RaftActorTest.fromObject(getState()).toByteArray()), actorRef); - } catch (Exception e) { - LOG.error("createSnapshot failed", e); + @SuppressWarnings("checkstyle:IllegalCatch") + public void createSnapshot(final ActorRef actorRef, final Optional installSnapshotStream) { + MockSnapshotState snapshotState = new MockSnapshotState(new ArrayList<>(getState())); + if (installSnapshotStream.isPresent()) { + SerializationUtils.serialize(snapshotState, installSnapshotStream.get()); } + + actorRef.tell(new CaptureSnapshotReply(snapshotState, installSnapshotStream), actorRef); } - static Props props(Map peerAddresses, RaftActorContext fromContext) { + static Props props(final Map peerAddresses, final RaftActorContext fromContext) { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); configParams.setElectionTimeoutFactor(10); @@ -1539,12 +1589,13 @@ 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); + public MockNewFollowerRaftActor(final ConfigParams config, final ActorRef collectorActor) { + super(NEW_SERVER_ID, Maps.newHashMap(), Optional.of(config), NO_PERSISTENCE, + collectorActor); setPersistence(false); } - static Props props(ConfigParams config, TestActorRef collectorActor) { + static Props props(final ConfigParams config, final ActorRef collectorActor) { return Props.create(MockNewFollowerRaftActor.class, config, collectorActor); } }