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=3bf328bc167a001c64b33c3e905d1e5fe30f9697;hb=refs%2Fchanges%2F11%2F32511%2F8;hp=da85c67c06ef4f4e2a56cd6901d763648e3be9b2;hpb=482caffb6cc503ab1d156109a5dba8c2fe675f3d;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 da85c67c06..3bf328bc16 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,6 +8,8 @@ package org.opendaylight.controller.cluster.raft; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; 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; @@ -19,6 +21,7 @@ import akka.dispatch.Dispatchers; import akka.testkit.JavaTestKit; import akka.testkit.TestActorRef; import com.google.common.base.Optional; +import com.google.common.base.Stopwatch; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; import com.google.common.collect.Sets; @@ -39,13 +42,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.ChangeServersVotingStatus; 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.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; @@ -66,6 +71,7 @@ import scala.concurrent.duration.FiniteDuration; public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { static final String LEADER_ID = "leader"; static final String FOLLOWER_ID = "follower"; + static final String FOLLOWER_ID2 = "follower2"; static final String NEW_SERVER_ID = "new-server"; static final String NEW_SERVER_ID2 = "new-server2"; private static final Logger LOG = LoggerFactory.getLogger(RaftActorServerConfigurationSupportTest.class); @@ -151,7 +157,7 @@ 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()); // Verify ServerConfigurationPayload entry in leader's log @@ -234,7 +240,7 @@ 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()); // Verify ServerConfigurationPayload entry in leader's log @@ -275,7 +281,7 @@ 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()); // Verify ServerConfigurationPayload entry in leader's log @@ -312,7 +318,7 @@ 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()); @@ -408,7 +414,7 @@ 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); @@ -484,7 +490,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()); @@ -589,18 +595,49 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); - newFollowerRaftActor.underlyingActor().setDropMessageOfType(AppendEntries.class); + 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()); + 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 @@ -642,16 +679,23 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { @Test public void testOnApplyState() { - RaftActorServerConfigurationSupport support = new RaftActorServerConfigurationSupport(new MockRaftActorContext()); + 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), null, ActorRef.noSender()); + 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), null, ActorRef.noSender()); + handled = support.handleMessage(new ApplyState(null, null, nonServerConfigEntry), ActorRef.noSender()); assertEquals("Message handled", false, handled); } @@ -686,48 +730,71 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testRemoveServerSelf() { - RaftActorContext initialActorContext = new MockRaftActorContext(); + public void testRemoveServerForwardToLeader() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - TestActorRef leaderActor = actorFactory.createTestActor( - MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), - initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + TestActorRef leaderActor = actorFactory.createTestActor( + MessageCollectorActor.props().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()); + 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 testRemoveServerForwardToLeader() { + 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(), + MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActorPath), initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(LEADER_ID)); - TestActorRef followerRaftActor = actorFactory.createTestActor( - MockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString()), + 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()), - actorFactory.generateActorId(FOLLOWER_ID)); + followerActorId); + TestActorRef collector = + actorFactory.createTestActor(MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); - followerRaftActor.tell(new AppendEntries(1, LEADER_ID, 0, 1, Collections.emptyList(), - -1, -1, (short) 0), leaderActor); + followerRaftActor.underlyingActor().setCollectorActor(collector); - followerRaftActor.tell(new RemoveServer(LEADER_ID), testKit.getRef()); + leaderActor.tell(new RemoveServer(FOLLOWER_ID), testKit.getRef()); RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); - assertEquals("getStatus", ServerChangeStatus.NOT_SUPPORTED, removeServerReply.getStatus()); + 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 testRemoveServer() { + public void testRemoveServerLeader() { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); configParams.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName()); @@ -748,37 +815,201 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { configParams, NO_PERSISTENCE).withDispatcher(Dispatchers.DefaultDispatcherId()), followerActorId); - TestActorRef collector = - actorFactory.createTestActor(MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); + TestActorRef followerCollector = actorFactory.createTestActor(MessageCollectorActor.props(). + withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId("collector")); + followerRaftActor.underlyingActor().setCollectorActor(followerCollector); - followerRaftActor.underlyingActor().setCollectorActor(collector); + leaderActor.tell(new RemoveServer(LEADER_ID), testKit.getRef()); + RemoveServerReply removeServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), RemoveServerReply.class); + assertEquals("getStatus", ServerChangeStatus.OK, removeServerReply.getStatus()); - leaderActor.tell(new RemoveServer(FOLLOWER_ID), testKit.getRef()); + 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()); + } - assertEquals("getStatus", ServerChangeStatus.OK, removeServerReply.getStatus()); + @Test + public void testChangeServersVotingStatus() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + configParams.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName()); + + 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); + + 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 follower1RaftActor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), + FOLLOWER_ID2, follower2ActorPath), configParams, NO_PERSISTENCE). + withDispatcher(Dispatchers.DefaultDispatcherId()), follower1ActorId); + TestActorRef follower1Collector = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId("collector")); + follower1RaftActor.underlyingActor().setCollectorActor(follower1Collector); + + TestActorRef follower2RaftActor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID2, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), + FOLLOWER_ID, follower1ActorPath), configParams, NO_PERSISTENCE). + withDispatcher(Dispatchers.DefaultDispatcherId()), follower2ActorId); + TestActorRef follower2Collector = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId("collector")); + follower2RaftActor.underlyingActor().setCollectorActor(follower2Collector); + + // 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); + assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); final ApplyState applyState = MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); assertEquals(0L, applyState.getReplicatedLogEntry().getIndex()); - verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), votingServer(LEADER_ID)); + verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + votingServer(LEADER_ID), nonVotingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); - MessageCollectorActor.expectFirstMatching(collector, ServerRemoved.class); + MessageCollectorActor.expectFirstMatching(follower1Collector, ApplyState.class); + 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)); + + MessageCollectorActor.clearMessages(leaderCollector); + MessageCollectorActor.clearMessages(follower1Collector); + MessageCollectorActor.clearMessages(follower2Collector); + + // Send second ChangeServersVotingStatus message + + leaderActor.tell(new ChangeServersVotingStatus(ImmutableMap.of(FOLLOWER_ID, true)), testKit.getRef()); + reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); + + MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); + verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + 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)); + + MessageCollectorActor.expectFirstMatching(follower2Collector, ApplyState.class); + verifyServerConfigurationPayloadEntry(follower2RaftActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + votingServer(LEADER_ID), votingServer(FOLLOWER_ID), nonVotingServer(FOLLOWER_ID2)); } - private ServerInfo votingServer(String id) { + @Test + public void testChangeLeaderToNonVoting() { + DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); + configParams.setHeartBeatInterval(new FiniteDuration(500, TimeUnit.MILLISECONDS)); + + 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); + + 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 follower1RaftActor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), + FOLLOWER_ID2, follower2ActorPath), configParams, NO_PERSISTENCE). + withDispatcher(Dispatchers.DefaultDispatcherId()), follower1ActorId); + TestActorRef follower1Collector = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId("collector")); + follower1RaftActor.underlyingActor().setCollectorActor(follower1Collector); + + TestActorRef follower2RaftActor = actorFactory.createTestActor( + CollectingMockRaftActor.props(FOLLOWER_ID2, ImmutableMap.of(LEADER_ID, leaderActor.path().toString(), + FOLLOWER_ID, follower1ActorPath), configParams, NO_PERSISTENCE). + withDispatcher(Dispatchers.DefaultDispatcherId()), follower2ActorId); + TestActorRef follower2Collector = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId("collector")); + follower2RaftActor.underlyingActor().setCollectorActor(follower2Collector); + + // Send ChangeServersVotingStatus message + + leaderActor.tell(new ChangeServersVotingStatus(ImmutableMap.of(LEADER_ID, false)), testKit.getRef()); + ServerChangeReply reply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), ServerChangeReply.class); + assertEquals("getStatus", ServerChangeStatus.OK, reply.getStatus()); + + MessageCollectorActor.expectFirstMatching(leaderCollector, ApplyState.class); + verifyServerConfigurationPayloadEntry(leaderActor.underlyingActor().getRaftActorContext().getReplicatedLog(), + 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)); + + MessageCollectorActor.expectFirstMatching(follower2Collector, ApplyState.class); + 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()); + + MessageCollectorActor.expectMatching(leaderCollector, AppendEntries.class, 2); + } + + private void verifyRaftState(RaftState expState, RaftActor... raftActors) { + Stopwatch sw = Stopwatch.createStarted(); + while(sw.elapsed(TimeUnit.SECONDS) <= 5) { + for(RaftActor raftActor: raftActors) { + if(raftActor.getRaftState() == expState) { + return; + } + } + } + + fail("None of the RaftActors have state " + expState); + } + + 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) { - TestActorRef leaderCollectorActor = actorFactory.createTestActor( + return newCollectorActor(leaderRaftActor, LEADER_ID); + } + + private TestActorRef newCollectorActor(AbstractMockRaftActor raftActor, String id) { + TestActorRef collectorActor = actorFactory.createTestActor( MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(LEADER_ID + "Collector")); - leaderRaftActor.setCollectorActor(leaderCollectorActor); - return leaderCollectorActor; + actorFactory.generateActorId(id + "Collector")); + raftActor.setCollectorActor(collectorActor); + return collectorActor; } private static void verifyServerConfigurationPayloadEntry(ReplicatedLog log, ServerInfo... expected) {