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=16acb410fb169702ee82cc7908f3e0b61abc5eb2;hp=da7b4eddf4951db8890964c12531b4f1e8d22f1e;hb=8b3db9dc8b4ac15d0b8bc133966fd3521f4909da;hpb=af65c60b352ab06eb32294ccd9879f5e43585e31 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 da7b4eddf4..16acb410fb 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,10 @@ package org.opendaylight.controller.cluster.raft; import static org.junit.Assert.assertEquals; +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; @@ -33,14 +35,17 @@ import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload.Serve 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.InitiateCaptureSnapshot; 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.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.ServerChangeStatus; +import org.opendaylight.controller.cluster.raft.messages.UnInitializedFollowerSnapshotReply; import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy; import org.opendaylight.controller.cluster.raft.utils.ForwardMessageToBehaviorActor; import org.opendaylight.controller.cluster.raft.utils.InMemoryJournal; @@ -130,12 +135,14 @@ 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()); @@ -145,6 +152,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // 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()); @@ -154,10 +162,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)); @@ -175,6 +185,21 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { 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()); } @Test @@ -193,11 +218,14 @@ 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()); @@ -207,6 +235,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // 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()); @@ -228,9 +257,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(), @@ -240,6 +266,8 @@ 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); @@ -248,6 +276,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { // 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()); @@ -265,10 +295,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); @@ -279,6 +311,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); assertEquals("getLeaderHint", 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()); @@ -289,9 +322,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(), @@ -301,6 +331,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); @@ -314,12 +346,15 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { Object installSnapshot = expectFirstMatching(newFollowerCollectorActor, InstallSnapshot.class); + // Send a second AddServer - should get queued JavaTestKit testKit2 = new JavaTestKit(getSystem()); leaderActor.tell(new AddServer(NEW_SERVER_ID2, followerActor.path().toString(), false), testKit2.getRef()); + // Continue the first AddServer newFollowerRaftActorInstance.setDropMessageOfType(null); newFollowerRaftActor.tell(installSnapshot, leaderActor); + // Verify both complete successfully AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); @@ -328,6 +363,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()); @@ -336,20 +372,56 @@ 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()); } @Test - public void testAddServerWithInstallSnapshotTimeout() throws Exception { - newFollowerRaftActor.underlyingActor().setDropMessageOfType(InstallSnapshot.SERIALIZABLE_CLASS); + public void testAddServerWithPriorSnapshotInProgress() throws Exception { + 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 commit message for now to delay snapshot completion + leaderRaftActor.setDropMessageOfType(String.class); + + leaderActor.tell(new InitiateCaptureSnapshot(), leaderActor); + + String commitMsg = expectFirstMatching(leaderCollectorActor, String.class); + + leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); + leaderRaftActor.setDropMessageOfType(null); + leaderActor.tell(commitMsg, leaderActor); + + AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); + assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint()); + + 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()); + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(NEW_SERVER_ID)); + } + + @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(), @@ -358,13 +430,129 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(1); + // Drop commit message so the snapshot doesn't complete. + leaderRaftActor.setDropMessageOfType(String.class); + + leaderActor.tell(new InitiateCaptureSnapshot(), leaderActor); + leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.TIMEOUT, addServerReply.getStatus()); + assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); + } + + @Test + public void testAddServerWithLeaderChangeBeforePriorSnapshotComplete() throws Exception { + 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(); + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(100); + + TestActorRef leaderCollectorActor = newLeaderCollectorActor(leaderRaftActor); + + // Drop the commit message so the snapshot doesn't complete yet. + leaderRaftActor.setDropMessageOfType(String.class); + + leaderActor.tell(new InitiateCaptureSnapshot(), leaderActor); + + leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); + + String commitMsg = expectFirstMatching(leaderCollectorActor, String.class); + + // Change the leader behavior to follower + leaderActor.tell(new Follower(leaderActorContext), leaderActor); + + // Drop CaptureSnapshotReply in case install snapshot is incorrectly initiated after the prior + // snapshot completes. This will prevent the invalid snapshot from completing and fail the + // isCapturing assertion below. + leaderRaftActor.setDropMessageOfType(CaptureSnapshotReply.class); + + // 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); + + AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); + + assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); + assertEquals("isCapturing", false, leaderActorContext.getSnapshotManager().isCapturing()); + } + + @Test + public void testAddServerWithLeaderChangeDuringInstallSnapshot() throws Exception { + 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(); + + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(8); + + TestActorRef 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, + UnInitializedFollowerSnapshotReply.class); + + // Prevent election timeout when the leader switches to follower + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(100); + + // Change the leader behavior to follower + leaderActor.tell(new Follower(leaderActorContext), leaderActor); + + // Send the captured UnInitializedFollowerSnapshotReply - should be a no-op + leaderRaftActor.setDropMessageOfType(null); + leaderActor.tell(snapshotReply, leaderActor); + + AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); + + assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); + } + + @Test + public void testAddServerWithInstallSnapshotTimeout() throws Exception { + 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(); + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(1); + + // Drop the InstallSnapshot message so it times out + newFollowerRaftActor.underlyingActor().setDropMessageOfType(InstallSnapshot.SERIALIZABLE_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); + assertEquals("getStatus", ServerChangeStatus.TIMEOUT, addServerReply.getStatus()); + assertEquals("Leader peers size", 0, leaderActorContext.getPeerIds().size()); assertEquals("Leader followers size", 0, ((AbstractLeader)leaderRaftActor.getCurrentBehavior()).getFollowerIds().size()); @@ -386,6 +574,47 @@ 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(); + + newFollowerRaftActor.underlyingActor().setDropMessageOfType(AppendEntries.class); + + leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); + + AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); + assertEquals("getLeaderHint", LEADER_ID, addServerReply.getLeaderHint()); + + // Verify ServerConfigurationPayload entry in leader's log + + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(NEW_SERVER_ID)); + } + + @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(); @@ -408,6 +637,21 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { expectFirstMatching(leaderActor, AddServer.class); } + @Test + public void testOnApplyState() { + RaftActorServerConfigurationSupport support = new RaftActorServerConfigurationSupport(new MockRaftActorContext()); + + ReplicatedLogEntry serverConfigEntry = new MockRaftActorContext.MockReplicatedLogEntry(1, 1, + new ServerConfigurationPayload(Collections.emptyList())); + boolean handled = support.handleMessage(new ApplyState(null, null, serverConfigEntry), null, 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()); + assertEquals("Message handled", false, handled); + } + private ServerInfo votingServer(String id) { return new ServerInfo(id, true); } @@ -416,6 +660,14 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { return new ServerInfo(id, false); } + private TestActorRef newLeaderCollectorActor(MockLeaderRaftActor leaderRaftActor) { + TestActorRef leaderCollectorActor = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID + "Collector")); + leaderRaftActor.setCollectorActor(leaderCollectorActor); + return leaderCollectorActor; + } + 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()); @@ -429,20 +681,45 @@ 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 { + private volatile TestActorRef collectorActor; + private volatile Class dropMessageOfType; + + AbstractMockRaftActor(String id, Map peerAddresses, Optional config, + DataPersistenceProvider dataPersistenceProvider, TestActorRef collectorActor) { + super(id, peerAddresses, config, dataPersistenceProvider); + this.collectorActor = collectorActor; + } + + void setDropMessageOfType(Class dropMessageOfType) { + this.dropMessageOfType = dropMessageOfType; + } + + void setCollectorActor(TestActorRef collectorActor) { + this.collectorActor = collectorActor; + } + + @Override + public void handleCommand(Object message) { + if(dropMessageOfType == null || !dropMessageOfType.equals(message.getClass())) { + super.handleCommand(message); + } + + if(collectorActor != null) { + collectorActor.tell(message, getSender()); + } + } } - public static class MockLeaderRaftActor extends MockRaftActor { + public static class MockLeaderRaftActor extends AbstractMockRaftActor { public MockLeaderRaftActor(Map peerAddresses, ConfigParams config, RaftActorContext fromContext) { - super(LEADER_ID, peerAddresses, Optional.of(config), NO_PERSISTENCE); + super(LEADER_ID, peerAddresses, Optional.of(config), NO_PERSISTENCE, null); + setPersistence(false); RaftActorContext context = getRaftActorContext(); for(int i = 0; i < fromContext.getReplicatedLog().size(); i++) { @@ -480,26 +757,10 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } } - public static class MockNewFollowerRaftActor extends MockRaftActor { - private final TestActorRef collectorActor; - private volatile Class dropMessageOfType; - + public static class MockNewFollowerRaftActor extends AbstractMockRaftActor { public MockNewFollowerRaftActor(ConfigParams config, TestActorRef collectorActor) { - super(NEW_SERVER_ID, Maps.newHashMap(), Optional.of(config), null); - this.collectorActor = collectorActor; - } - - void setDropMessageOfType(Class dropMessageOfType) { - this.dropMessageOfType = dropMessageOfType; - } - - @Override - public void handleCommand(Object message) { - if(dropMessageOfType == null || !dropMessageOfType.equals(message.getClass())) { - super.handleCommand(message); - } - - collectorActor.tell(message, getSender()); + super(NEW_SERVER_ID, Maps.newHashMap(), Optional.of(config), null, collectorActor); + setPersistence(false); } static Props props(ConfigParams config, TestActorRef collectorActor) {