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=77f40cd7c15038d3ac92ffbcd27fdf0bb9652ee3;hb=8b3db9dc8b4ac15d0b8bc133966fd3521f4909da;hpb=3bc363a69d6d48709f7fd741ef018ecd75b8f99b 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 77f40cd7c1..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; @@ -19,7 +21,9 @@ import akka.testkit.TestActorRef; import com.google.common.base.Optional; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Maps; +import com.google.common.collect.Sets; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import org.junit.After; @@ -27,15 +31,25 @@ import org.junit.Before; import org.junit.Test; import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.NonPersistentDataProvider; +import org.opendaylight.controller.cluster.raft.ServerConfigurationPayload.ServerInfo; +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; +import org.opendaylight.controller.cluster.raft.utils.InMemorySnapshotStore; import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -50,6 +64,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { static final String LEADER_ID = "leader"; static final String FOLLOWER_ID = "follower"; 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); private static final DataPersistenceProvider NO_PERSISTENCE = new NonPersistentDataProvider(); @@ -59,23 +74,39 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { Props.create(ForwardMessageToBehaviorActor.class).withDispatcher(Dispatchers.DefaultDispatcherId()), actorFactory.generateActorId(FOLLOWER_ID)); - private final TestActorRef newServerActor = actorFactory.createTestActor( - Props.create(ForwardMessageToBehaviorActor.class).withDispatcher(Dispatchers.DefaultDispatcherId()), - actorFactory.generateActorId(NEW_SERVER_ID)); + private TestActorRef newFollowerRaftActor; + private TestActorRef newFollowerCollectorActor; + private RaftActorContext newFollowerActorContext; - private RaftActorContext newServerActorContext; private final JavaTestKit testKit = new JavaTestKit(getSystem()); @Before public void setup() { + InMemoryJournal.clear(); + InMemorySnapshotStore.clear(); + + DefaultConfigParamsImpl configParams = newFollowerConfigParams(); + + newFollowerCollectorActor = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(NEW_SERVER_ID + "Collector")); + newFollowerRaftActor = actorFactory.createTestActor(MockNewFollowerRaftActor.props( + configParams, newFollowerCollectorActor).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(NEW_SERVER_ID)); + + try { + newFollowerActorContext = newFollowerRaftActor.underlyingActor().getRaftActorContext(); + } catch (Exception e) { + newFollowerActorContext = newFollowerRaftActor.underlyingActor().getRaftActorContext(); + } + } + + private static DefaultConfigParamsImpl newFollowerConfigParams() { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); configParams.setElectionTimeoutFactor(100000); configParams.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName()); - newServerActorContext = new RaftActorContextImpl(newServerActor, newServerActor.underlyingActor().getContext(), - NEW_SERVER_ID, new ElectionTermImpl(NO_PERSISTENCE, NEW_SERVER_ID, LOG), -1, -1, - Maps.newHashMap(), configParams, NO_PERSISTENCE, LOG); - newServerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); + return configParams; } @After @@ -84,19 +115,16 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithFollower() throws Exception { + public void testAddServerWithExistingFollower() throws Exception { RaftActorContext followerActorContext = newFollowerContext(FOLLOWER_ID, followerActor); followerActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries( 0, 3, 1).build()); - followerActorContext.setCommitIndex(3); - followerActorContext.setLastApplied(3); + followerActorContext.setCommitIndex(2); + followerActorContext.setLastApplied(2); Follower follower = new Follower(followerActorContext); followerActor.underlyingActor().setBehavior(follower); - Follower newServer = new Follower(newServerActorContext); - newServerActor.underlyingActor().setBehavior(newServer); - TestActorRef leaderActor = actorFactory.createTestActor( MockLeaderRaftActor.props(ImmutableMap.of(FOLLOWER_ID, followerActor.path().toString()), followerActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), @@ -107,31 +135,427 @@ 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()); + + // Verify ServerConfigurationPayload entry in leader's log - leaderActor.tell(new AddServer(NEW_SERVER_ID, newServerActor.path().toString(), true), testKit.getRef()); + expectFirstMatching(leaderCollectorActor, ApplyState.class); + RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + assertEquals("Leader journal last index", 3, leaderActorContext.getReplicatedLog().lastIndex()); + assertEquals("Leader commit index", 3, leaderActorContext.getCommitIndex()); + assertEquals("Leader last applied index", 3, leaderActorContext.getLastApplied()); + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(FOLLOWER_ID), votingServer(NEW_SERVER_ID)); - // leader should install snapshot - capture and verify ApplySnapshot contents -// ApplySnapshot applySnapshot = expectFirstMatching(followerActor, ApplySnapshot.class); -// List snapshotState = (List) MockRaftActor.toObject(applySnapshot.getSnapshot().getState()); -// assertEquals("Snapshot state", snapshotState, leaderRaftActor.getState()); + // Verify ServerConfigurationPayload entry in both followers - // leader should replicate new server config to both followers -// expectFirstMatching(followerActor, AppendEntries.class); -// expectFirstMatching(newServerActor, AppendEntries.class); + 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)); - // verify ServerConfigurationPayload entry in leader's log -// RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); -// assertEquals("Leader journal log size", 4, leaderActorContext.getReplicatedLog().size()); -// assertEquals("Leader journal last index", 3, leaderActorContext.getReplicatedLog().lastIndex()); -// ReplicatedLogEntry logEntry = leaderActorContext.getReplicatedLog().get( -// leaderActorContext.getReplicatedLog().lastIndex()); - // verify logEntry contents + 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)); + + // Verify new server config was applied in both followers + + assertEquals("Follower peers", Sets.newHashSet(LEADER_ID, NEW_SERVER_ID), followerActorContext.getPeerIds()); + + assertEquals("New follower peers", Sets.newHashSet(LEADER_ID, FOLLOWER_ID), newFollowerActorContext.getPeerIds()); + + expectFirstMatching(newFollowerCollectorActor, ApplyState.class); + expectFirstMatching(followerActor, ApplyState.class); + + assertEquals("Follower commit index", 3, followerActorContext.getCommitIndex()); + assertEquals("Follower last applied index", 3, followerActorContext.getLastApplied()); + assertEquals("New follower commit index", 3, newFollowerActorContext.getCommitIndex()); + assertEquals("New follower last applied index", 3, newFollowerActorContext.getLastApplied()); + + List persistedLogEntries = InMemoryJournal.get(LEADER_ID, ReplicatedLogImplEntry.class); + assertEquals("Leader ReplicatedLogImplEntry entries", 1, persistedLogEntries.size()); + ReplicatedLogImplEntry logEntry = persistedLogEntries.get(0); + assertEquals("Leader ReplicatedLogImplEntry getTerm", 1, logEntry.getTerm()); + assertEquals("Leader ReplicatedLogImplEntry getIndex", 3, logEntry.getIndex()); + assertEquals("Leader ReplicatedLogImplEntry getData", ServerConfigurationPayload.class, logEntry.getData().getClass()); + + persistedLogEntries = InMemoryJournal.get(NEW_SERVER_ID, ReplicatedLogImplEntry.class); + assertEquals("New follower ReplicatedLogImplEntry entries", 1, persistedLogEntries.size()); + logEntry = persistedLogEntries.get(0); + assertEquals("New follower ReplicatedLogImplEntry getTerm", 1, logEntry.getTerm()); + assertEquals("New follower ReplicatedLogImplEntry getIndex", 3, logEntry.getIndex()); + assertEquals("New follower ReplicatedLogImplEntry getData", ServerConfigurationPayload.class, + logEntry.getData().getClass()); + } + + @Test + public void testAddServerWithNoExistingFollower() throws Exception { + RaftActorContext initialActorContext = new MockRaftActorContext(); + initialActorContext.setCommitIndex(1); + initialActorContext.setLastApplied(1); + initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries( + 0, 2, 1).build()); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(ImmutableMap.of(), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); + RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); - // Also verify ServerConfigurationPayload entry in both followers + 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()); + + // 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()); + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(NEW_SERVER_ID)); + + // Verify ServerConfigurationPayload entry in the new follower + + expectFirstMatching(newFollowerCollectorActor, ApplyState.class); + assertEquals("New follower journal last index", 2, newFollowerActorContext.getReplicatedLog().lastIndex()); + verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(NEW_SERVER_ID)); + + // Verify new server config was applied in the new follower + + assertEquals("New follower peers", Sets.newHashSet(LEADER_ID), newFollowerActorContext.getPeerIds()); + } + + @Test + public void testAddServersAsNonVoting() 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); + + 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()); + + // 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), + nonVotingServer(NEW_SERVER_ID)); + + // Verify ServerConfigurationPayload entry in the new follower + + expectFirstMatching(newFollowerCollectorActor, ApplyState.class); + assertEquals("New follower journal last index", 0, newFollowerActorContext.getReplicatedLog().lastIndex()); + verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), votingServer(LEADER_ID), + nonVotingServer(NEW_SERVER_ID)); + + // Verify new server config was applied in the new follower + + assertEquals("New follower peers", Sets.newHashSet(LEADER_ID), newFollowerActorContext.getPeerIds()); + + 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); + + leaderActor.tell(new AddServer(NEW_SERVER_ID2, followerActor.path().toString(), false), testKit.getRef()); + + addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + 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()); + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), + votingServer(LEADER_ID), nonVotingServer(NEW_SERVER_ID), nonVotingServer(NEW_SERVER_ID2)); + } + + @Test + public void testAddServerWithOperationInProgress() 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); + + RaftActorContext follower2ActorContext = newFollowerContext(NEW_SERVER_ID2, followerActor); + Follower newFollower2 = new Follower(follower2ActorContext); + followerActor.underlyingActor().setBehavior(newFollower2); + + MockNewFollowerRaftActor newFollowerRaftActorInstance = newFollowerRaftActor.underlyingActor(); + newFollowerRaftActorInstance.setDropMessageOfType(InstallSnapshot.SERIALIZABLE_CLASS); + + leaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); + + // Wait for leader's install snapshot and capture it + + 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()); + + addServerReply = testKit2.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); + assertEquals("getStatus", ServerChangeStatus.OK, addServerReply.getStatus()); + + // 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()); + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), + votingServer(LEADER_ID), votingServer(NEW_SERVER_ID), nonVotingServer(NEW_SERVER_ID2)); + + // Verify ServerConfigurationPayload entry in the new follower + + expectMatching(newFollowerCollectorActor, ApplyState.class, 2); + assertEquals("New follower peers", Sets.newHashSet(LEADER_ID, NEW_SERVER_ID2), + newFollowerActorContext.getPeerIds()); + } + + @Test + 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(); + + 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 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()); } @Test @@ -145,11 +569,52 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { actorFactory.generateActorId(LEADER_ID)); noLeaderActor.underlyingActor().waitForInitializeBehaviorComplete(); - noLeaderActor.tell(new AddServer(NEW_SERVER_ID, newServerActor.path().toString(), true), testKit.getRef()); + noLeaderActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); AddServerReply addServerReply = testKit.expectMsgClass(JavaTestKit.duration("5 seconds"), AddServerReply.class); assertEquals("getStatus", ServerChangeStatus.NO_LEADER, addServerReply.getStatus()); } + @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(); @@ -168,25 +633,93 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { followerRaftActor.tell(new AppendEntries(1, LEADER_ID, 0, 1, Collections.emptyList(), -1, -1, (short)0), leaderActor); - followerRaftActor.tell(new AddServer(NEW_SERVER_ID, newServerActor.path().toString(), true), testKit.getRef()); + followerRaftActor.tell(new AddServer(NEW_SERVER_ID, newFollowerRaftActor.path().toString(), true), testKit.getRef()); expectFirstMatching(leaderActor, AddServer.class); } - private RaftActorContext newFollowerContext(String id, TestActorRef actor) { + @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); + } + + private ServerInfo nonVotingServer(String id) { + 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()); + ServerConfigurationPayload payload = (ServerConfigurationPayload)logEntry.getData(); + assertEquals("getNewServerConfig", Sets.newHashSet(expected), Sets.newHashSet(payload.getServerConfig())); + } + + private static RaftActorContext newFollowerContext(String id, TestActorRef actor) { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); configParams.setElectionTimeoutFactor(100000); - RaftActorContext followerActorContext = new RaftActorContextImpl(actor, actor.underlyingActor().getContext(), - id, new ElectionTermImpl(NO_PERSISTENCE, id, LOG), -1, -1, - ImmutableMap.of(LEADER_ID, ""), configParams, NO_PERSISTENCE, LOG); + ElectionTermImpl termInfo = new ElectionTermImpl(NO_PERSISTENCE, id, LOG); + termInfo.update(1, LEADER_ID); + 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); + } - return followerActorContext; + 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++) { @@ -197,6 +730,8 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { context.setCommitIndex(fromContext.getCommitIndex()); context.setLastApplied(fromContext.getLastApplied()); + context.getTermInformation().update(fromContext.getTermInformation().getCurrentTerm(), + fromContext.getTermInformation().getVotedFor()); } @Override @@ -216,9 +751,20 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { static Props props(Map peerAddresses, RaftActorContext fromContext) { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); - configParams.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - configParams.setElectionTimeoutFactor(100000); + configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); + configParams.setElectionTimeoutFactor(10); return Props.create(MockLeaderRaftActor.class, peerAddresses, configParams, fromContext); } } + + public static class MockNewFollowerRaftActor extends AbstractMockRaftActor { + public MockNewFollowerRaftActor(ConfigParams config, TestActorRef collectorActor) { + super(NEW_SERVER_ID, Maps.newHashMap(), Optional.of(config), null, collectorActor); + setPersistence(false); + } + + static Props props(ConfigParams config, TestActorRef collectorActor) { + return Props.create(MockNewFollowerRaftActor.class, config, collectorActor); + } + } }