X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-akka-raft%2Fsrc%2Ftest%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fraft%2FRaftActorServerConfigurationSupportTest.java;h=36f6fe7b1a4647d0db5e0b83482234be675321a0;hb=24a5bafd22b83c4d838b7c3fc5225934fe969561;hp=3d6f7f414ce7b1f1fc03e876fb7b35912656f23a;hpb=93ccb43ad0f5c78337f19884a51e2bd479cc46fd;p=controller.git 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 3d6f7f414c..36f6fe7b1a 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 @@ -8,8 +8,11 @@ package org.opendaylight.controller.cluster.raft; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor.assertNoneMatching; 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; @@ -37,12 +40,15 @@ import org.opendaylight.controller.cluster.raft.base.messages.InitiateCaptureSna import org.opendaylight.controller.cluster.raft.behaviors.AbstractLeader; import org.opendaylight.controller.cluster.raft.behaviors.Follower; import org.opendaylight.controller.cluster.raft.behaviors.Leader; +import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior; import org.opendaylight.controller.cluster.raft.messages.AddServer; import org.opendaylight.controller.cluster.raft.messages.AddServerReply; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; -import org.opendaylight.controller.cluster.raft.messages.FollowerCatchUpTimeout; import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot; +import org.opendaylight.controller.cluster.raft.messages.RemoveServer; +import org.opendaylight.controller.cluster.raft.messages.RemoveServerReply; 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.policy.DisableElectionsRaftPolicy; import org.opendaylight.controller.cluster.raft.utils.ForwardMessageToBehaviorActor; @@ -133,21 +139,24 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { clearMessages(followerActor); MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); + TestActorRef 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()); assertEquals("Snapshot state", snapshotState, leaderRaftActor.getState()); AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); - assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint()); + assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); // Verify ServerConfigurationPayload entry in leader's log + expectFirstMatching(leaderCollectorActor, ApplyState.class); RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); assertEquals("Leader journal last index", 3, leaderActorContext.getReplicatedLog().lastIndex()); assertEquals("Leader commit index", 3, leaderActorContext.getCommitIndex()); @@ -157,10 +166,12 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // Verify ServerConfigurationPayload entry in both followers + expectFirstMatching(followerActor, ApplyState.class); assertEquals("Follower journal last index", 3, followerActorContext.getReplicatedLog().lastIndex()); verifyServerConfigurationPayloadEntry(followerActorContext.getReplicatedLog(), votingServer(LEADER_ID), votingServer(FOLLOWER_ID), votingServer(NEW_SERVER_ID)); + expectFirstMatching(newFollowerCollectorActor, ApplyState.class); assertEquals("New follower journal last index", 3, newFollowerActorContext.getReplicatedLog().lastIndex()); verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), votingServer(LEADER_ID), votingServer(FOLLOWER_ID), votingServer(NEW_SERVER_ID)); @@ -211,20 +222,24 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + TestActorRef 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()); assertEquals("Snapshot state", snapshotState, leaderRaftActor.getState()); AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); - assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint()); + assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); // Verify ServerConfigurationPayload entry in leader's log + expectFirstMatching(leaderCollectorActor, ApplyState.class); assertEquals("Leader journal last index", 2, leaderActorContext.getReplicatedLog().lastIndex()); assertEquals("Leader commit index", 2, leaderActorContext.getCommitIndex()); assertEquals("Leader last applied index", 2, leaderActorContext.getLastApplied()); @@ -246,9 +261,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testAddServersAsNonVoting() throws Exception { RaftActorContext initialActorContext = new MockRaftActorContext(); - initialActorContext.setCommitIndex(-1); - initialActorContext.setLastApplied(-1); - initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(), @@ -258,14 +270,18 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + TestActorRef 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); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); - assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint()); + assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); // Verify ServerConfigurationPayload entry in leader's log + expectFirstMatching(leaderCollectorActor, ApplyState.class); + assertEquals("Leader journal last index", 0, leaderActorContext.getReplicatedLog().lastIndex()); assertEquals("Leader commit index", 0, leaderActorContext.getCommitIndex()); assertEquals("Leader last applied index", 0, leaderActorContext.getLastApplied()); @@ -283,10 +299,12 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { assertEquals("New follower peers", Sets.newHashSet(LEADER_ID), newFollowerActorContext.getPeerIds()); - MessageCollectorActor.assertNoneMatching(newFollowerCollectorActor, InstallSnapshot.class, 500); + assertNoneMatching(newFollowerCollectorActor, InstallSnapshot.class, 500); // Add another non-voting server. + clearMessages(leaderCollectorActor); + RaftActorContext follower2ActorContext = newFollowerContext(NEW_SERVER_ID2, followerActor); Follower newFollower2 = new Follower(follower2ActorContext); followerActor.underlyingActor().setBehavior(newFollower2); @@ -295,8 +313,9 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); - assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint()); + assertEquals("getLeaderHint", java.util.Optional.of(LEADER_ID), addServerReply.getLeaderHint()); + expectFirstMatching(leaderCollectorActor, ApplyState.class); assertEquals("Leader journal last index", 1, leaderActorContext.getReplicatedLog().lastIndex()); assertEquals("Leader commit index", 1, leaderActorContext.getCommitIndex()); assertEquals("Leader last applied index", 1, leaderActorContext.getLastApplied()); @@ -307,9 +326,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testAddServerWithOperationInProgress() throws Exception { RaftActorContext initialActorContext = new MockRaftActorContext(); - initialActorContext.setCommitIndex(-1); - initialActorContext.setLastApplied(-1); - initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(), @@ -319,6 +335,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + RaftActorContext follower2ActorContext = newFollowerContext(NEW_SERVER_ID2, followerActor); Follower newFollower2 = new Follower(follower2ActorContext); followerActor.underlyingActor().setBehavior(newFollower2); @@ -330,7 +348,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // Wait for leader's install snapshot and capture it - Object installSnapshot = expectFirstMatching(newFollowerCollectorActor, InstallSnapshot.class); + InstallSnapshot installSnapshot = expectFirstMatching(newFollowerCollectorActor, InstallSnapshot.class); // Send a second AddServer - should get queued JavaTestKit testKit2 = new JavaTestKit(getSystem()); @@ -349,6 +367,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // Verify ServerConfigurationPayload entries in leader's log + expectMatching(leaderCollectorActor, ApplyState.class, 2); assertEquals("Leader journal last index", 1, leaderActorContext.getReplicatedLog().lastIndex()); assertEquals("Leader commit index", 1, leaderActorContext.getCommitIndex()); assertEquals("Leader last applied index", 1, leaderActorContext.getLastApplied()); @@ -357,8 +376,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // Verify ServerConfigurationPayload entry in the new follower - MessageCollectorActor.expectMatching(newFollowerCollectorActor, ApplyState.class, 2); - + expectMatching(newFollowerCollectorActor, ApplyState.class, 2); assertEquals("New follower peers", Sets.newHashSet(LEADER_ID, NEW_SERVER_ID2), newFollowerActorContext.getPeerIds()); } @@ -366,9 +384,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testAddServerWithPriorSnapshotInProgress() throws Exception { RaftActorContext initialActorContext = new MockRaftActorContext(); - initialActorContext.setCommitIndex(-1); - initialActorContext.setLastApplied(-1); - initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(), @@ -378,10 +393,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); - TestActorRef leaderCollectorActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(LEADER_ID + "Collector")); - leaderRaftActor.setCollectorActor(leaderCollectorActor); + TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop commit message for now to delay snapshot completion leaderRaftActor.setDropMessageOfType(String.class); @@ -397,12 +409,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); - assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint()); + assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); expectFirstMatching(newFollowerCollectorActor, ApplySnapshot.class); // Verify ServerConfigurationPayload entry in leader's log + expectFirstMatching(leaderCollectorActor, ApplyState.class); assertEquals("Leader journal last index", 0, leaderActorContext.getReplicatedLog().lastIndex()); assertEquals("Leader commit index", 0, leaderActorContext.getCommitIndex()); assertEquals("Leader last applied index", 0, leaderActorContext.getLastApplied()); @@ -413,9 +426,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testAddServerWithPriorSnapshotCompleteTimeout() throws Exception { RaftActorContext initialActorContext = new MockRaftActorContext(); - initialActorContext.setCommitIndex(-1); - initialActorContext.setLastApplied(-1); - initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(), @@ -427,11 +437,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(1); - TestActorRef leaderCollectorActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(LEADER_ID + "Collector")); - leaderRaftActor.setCollectorActor(leaderCollectorActor); - // Drop commit message so the snapshot doesn't complete. leaderRaftActor.setDropMessageOfType(String.class); @@ -448,9 +453,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testAddServerWithLeaderChangeBeforePriorSnapshotComplete() throws Exception { RaftActorContext initialActorContext = new MockRaftActorContext(); - initialActorContext.setCommitIndex(-1); - initialActorContext.setLastApplied(-1); - initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(), @@ -461,10 +463,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(100); - TestActorRef leaderCollectorActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(LEADER_ID + "Collector")); - leaderRaftActor.setCollectorActor(leaderCollectorActor); + TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop the commit message so the snapshot doesn't complete yet. leaderRaftActor.setDropMessageOfType(String.class); @@ -486,7 +485,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // Complete the prior snapshot - this should be a no-op b/c it's no longer the leader leaderActor.tell(commitMsg, leaderActor); - leaderActor.tell(new FollowerCatchUpTimeout(NEW_SERVER_ID), leaderActor); + leaderActor.tell(new RaftActorServerConfigurationSupport.ServerOperationTimeout(NEW_SERVER_ID), leaderActor); AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); @@ -498,9 +497,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testAddServerWithLeaderChangeDuringInstallSnapshot() throws Exception { RaftActorContext initialActorContext = new MockRaftActorContext(); - initialActorContext.setCommitIndex(-1); - initialActorContext.setLastApplied(-1); - initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(), @@ -512,10 +508,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(8); - TestActorRef leaderCollectorActor = actorFactory.createTestActor( - MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(LEADER_ID + "Collector")); - leaderRaftActor.setCollectorActor(leaderCollectorActor); + TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); // Drop the UnInitializedFollowerSnapshotReply to delay it. leaderRaftActor.setDropMessageOfType(UnInitializedFollowerSnapshotReply.class); @@ -544,9 +537,6 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testAddServerWithInstallSnapshotTimeout() throws Exception { RaftActorContext initialActorContext = new MockRaftActorContext(); - initialActorContext.setCommitIndex(-1); - initialActorContext.setLastApplied(-1); - initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(), @@ -578,8 +568,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); TestActorRef noLeaderActor = actorFactory.createTestActor( - MockRaftActor.props(LEADER_ID, ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), - Optional.of(configParams), NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), + MockRaftActor.props(LEADER_ID, ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), + configParams, NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); noLeaderActor.underlyingActor().waitForInitializeBehaviorComplete(); @@ -588,6 +578,78 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); } + @Test + public void testAddServerWithNoConsensusReached() { + RaftActorContext initialActorContext = new MockRaftActorContext(); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(ImmutableMap.of(), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); + RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + + TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + + // Drop UnInitializedFollowerSnapshotReply initially + leaderRaftActor.setDropMessageOfType(UnInitializedFollowerSnapshotReply.class); + + MockNewFollowerRaftActor newFollowerRaftActorInstance = newFollowerRaftActor.underlyingActor(); + TestActorRef newFollowerCollectorActor = + newCollectorActor(newFollowerRaftActorInstance, NEW_SERVER_ID); + + // Drop AppendEntries to the new follower so consensus isn't reached + newFollowerRaftActorInstance.setDropMessageOfType(AppendEntries.class); + + leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); + + // Capture the UnInitializedFollowerSnapshotReply + Object snapshotReply = expectFirstMatching(leaderCollectorActor, UnInitializedFollowerSnapshotReply.class); + + // Send the UnInitializedFollowerSnapshotReply to resume the first request + leaderRaftActor.setDropMessageOfType(null); + leaderActor.tell(snapshotReply, leaderActor); + + expectFirstMatching(newFollowerCollectorActor, AppendEntries.class); + + // Send a second AddServer + 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); + assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); + assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint().get()); + + // Verify ServerConfigurationPayload entry in leader's log + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), votingServer(LEADER_ID), + 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); + 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); + assertEquals("getStatus", ServerChangeStatus.PRIOR_REQUEST_CONSENSUS_TIMEOUT, addServerReply.getStatus()); + } + + @Test + public void testAddServerWithExistingServer() { + RaftActorContext initialActorContext = new MockRaftActorContext(); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + leaderActor.tell(new AddServer(FOLLOWER_ID, followerActor.path().toString(), true), testKit.getRef()); + + AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + assertEquals("getStatus", ServerChangeStatus.ALREADY_EXISTS, addServerReply.getStatus()); + } + @Test public void testAddServerForwardedToLeader() { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); @@ -598,8 +660,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); TestActorRef followerRaftActor = actorFactory.createTestActor( - MockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), - Optional.of(configParams), NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), + MockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), + configParams, NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(FOLLOWER_ID)); followerRaftActor.underlyingActor().waitForInitializeBehaviorComplete(); @@ -610,14 +672,192 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { expectFirstMatching(leaderActor, AddServer.class); } - private ServerInfo votingServer(String id) { + @Test + public void testOnApplyState() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + TestActorRef noLeaderActor = actorFactory.createTestActor( + MockRaftActor.props(LEADER_ID, ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), + configParams, NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + RaftActorServerConfigurationSupport support = new RaftActorServerConfigurationSupport(noLeaderActor.underlyingActor()); + + ReplicatedLogEntry serverConfigEntry = new MockRaftActorContext.MockReplicatedLogEntry(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, + new MockRaftActorContext.MockPayload("1")); + handled = support.handleMessage(new ApplyState(null, null, nonServerConfigEntry), ActorRef.noSender()); + assertEquals("Message handled", false, handled); + } + + @Test + public void testRemoveServerWithNoLeader() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockRaftActor.props(LEADER_ID, ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), + configParams, NO_PERSISTENCE).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); + assertEquals("getStatus", ServerChangeStatus.NO_LEADER, removeServerReply.getStatus()); + } + + @Test + public void testRemoveServerNonExistentServer() { + RaftActorContext initialActorContext = new MockRaftActorContext(); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + leaderActor.tell(new RemoveServer(NEW_SERVER_ID), testKit.getRef()); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + assertEquals("getStatus", ServerChangeStatus.DOES_NOT_EXIST, removeServerReply.getStatus()); + } + + @Test + public void testRemoveServerForwardToLeader() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + TestActorRef leaderActor = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + TestActorRef followerRaftActor = actorFactory.createTestActor( + MockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), + configParams, NO_PERSISTENCE).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 RemoveServer(FOLLOWER_ID), testKit.getRef()); + expectFirstMatching(leaderActor, RemoveServer.class); + } + + @Test + public void testRemoveServer() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + 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); + RaftActorContext initialActorContext = new MockRaftActorContext(); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActorPath), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + + TestActorRef followerRaftActor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), + configParams, NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), + followerActorId); + + TestActorRef collector = + actorFactory.createTestActor(MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); + + followerRaftActor.underlyingActor().setCollectorActor(collector); + + leaderActor.tell(new RemoveServer(FOLLOWER_ID), testKit.getRef()); + 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)); + + RaftActorBehavior currentBehavior = leaderActor.underlyingActor().getCurrentBehavior(); + assertTrue("Expected Leader", currentBehavior instanceof Leader); + assertEquals("Follower ids size", 0, ((Leader)currentBehavior).getFollowerIds().size()); + + MessageCollectorActor.expectFirstMatching(collector, ServerRemoved.class); + } + + @Test + public void testRemoveServerLeader() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + 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); + RaftActorContext initialActorContext = new MockRaftActorContext(); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActorPath), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + TestActorRef leaderCollector = newLeaderCollectorActor(leaderActor.underlyingActor()); + + TestActorRef followerRaftActor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), + configParams, NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), + followerActorId); + + TestActorRef followerCollector = actorFactory.createTestActor(MessageCollectorActor.props(). + withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); + followerRaftActor.underlyingActor().setCollectorActor(followerCollector); + + leaderActor.tell(new RemoveServer(LEADER_ID), testKit.getRef()); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + assertEquals("getStatus", ServerChangeStatus.OK, removeServerReply.getStatus()); + + final ApplyState applyState = MessageCollectorActor.expectFirstMatching(followerCollector, ApplyState.class); + assertEquals(0L, applyState.getReplicatedLogEntry().getIndex()); + verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + votingServer(FOLLOWER_ID)); + + MessageCollectorActor.expectFirstMatching(leaderCollector, ServerRemoved.class); + } + + @Test + public void testRemoveServerLeaderWithNoFollowers() { + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(Collections.emptyMap(), + new MockRaftActorContext()).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + leaderActor.tell(new RemoveServer(LEADER_ID), testKit.getRef()); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + assertEquals("getStatus", ServerChangeStatus.NOT_SUPPORTED, removeServerReply.getStatus()); + } + + private static ServerInfo votingServer(String id) { return new ServerInfo(id, true); } - private ServerInfo nonVotingServer(String id) { + private static ServerInfo nonVotingServer(String id) { return new ServerInfo(id, false); } + private TestActorRef newLeaderCollectorActor(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")); + raftActor.setCollectorActor(collectorActor); + return collectorActor; + } + private static void verifyServerConfigurationPayloadEntry(ReplicatedLog log, ServerInfo... expected) { ReplicatedLogEntry logEntry = log.get(log.lastIndex()); assertEquals("Last log entry payload class", ServerConfigurationPayload.class, logEntry.getData().getClass()); @@ -631,14 +871,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { configParams.setElectionTimeoutFactor(100000); ElectionTermImpl termInfo = new ElectionTermImpl(NO_PERSISTENCE, id, LOG); termInfo.update(1, LEADER_ID); - RaftActorContext followerActorContext = new RaftActorContextImpl(actor, actor.underlyingActor().getContext(), - id, termInfo, -1, -1, - ImmutableMap.of(LEADER_ID, ""), configParams, NO_PERSISTENCE, LOG); - followerActorContext.setCommitIndex(-1); - followerActorContext.setLastApplied(-1); - followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); - - return followerActorContext; + return new RaftActorContextImpl(actor, actor.underlyingActor().getContext(), + id, termInfo, -1, -1, ImmutableMap.of(LEADER_ID, ""), configParams, NO_PERSISTENCE, LOG); } static abstract class AbstractMockRaftActor extends MockRaftActor { @@ -647,7 +881,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { AbstractMockRaftActor(String id, Map peerAddresses, Optional config, DataPersistenceProvider dataPersistenceProvider, TestActorRef collectorActor) { - super(id, peerAddresses, config, dataPersistenceProvider); + super(builder().id(id).peerAddresses(peerAddresses).config(config.get()). + dataPersistenceProvider(dataPersistenceProvider)); this.collectorActor = collectorActor; } @@ -671,6 +906,20 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } } + public static class CollectingMockRaftActor extends AbstractMockRaftActor { + + CollectingMockRaftActor(String id, Map peerAddresses, Optional config, DataPersistenceProvider dataPersistenceProvider, TestActorRef collectorActor) { + super(id, peerAddresses, config, dataPersistenceProvider, collectorActor); + } + + public static Props props(final String id, final Map peerAddresses, + ConfigParams config, DataPersistenceProvider dataPersistenceProvider){ + + return Props.create(CollectingMockRaftActor.class, id, peerAddresses, Optional.of(config), dataPersistenceProvider, null); + } + + } + public static class MockLeaderRaftActor extends AbstractMockRaftActor { public MockLeaderRaftActor(Map peerAddresses, ConfigParams config, RaftActorContext fromContext) {