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=3d6f7f414ce7b1f1fc03e876fb7b35912656f23a;hb=refs%2Fchanges%2F52%2F28952%2F4;hp=1b717f147d25819b2de5430f7f98414770fa2464;hpb=5aa58404a8ee1ad053742780439823309360a3a1;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 1b717f147d..3d6f7f414c 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 @@ -23,24 +23,27 @@ import com.google.common.collect.Sets; import java.util.Collections; import java.util.List; import java.util.Map; -//import java.util.List; import java.util.concurrent.TimeUnit; import org.junit.After; 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; @@ -96,7 +99,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } } - private DefaultConfigParamsImpl newFollowerConfigParams() { + private static DefaultConfigParamsImpl newFollowerConfigParams() { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); configParams.setElectionTimeoutFactor(100000); @@ -149,15 +152,18 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { 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(), LEADER_ID, FOLLOWER_ID, NEW_SERVER_ID); + verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(FOLLOWER_ID), votingServer(NEW_SERVER_ID)); // Verify ServerConfigurationPayload entry in both followers assertEquals("Follower journal last index", 3, followerActorContext.getReplicatedLog().lastIndex()); - verifyServerConfigurationPayloadEntry(followerActorContext.getReplicatedLog(), LEADER_ID, FOLLOWER_ID, NEW_SERVER_ID); + verifyServerConfigurationPayloadEntry(followerActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(FOLLOWER_ID), votingServer(NEW_SERVER_ID)); assertEquals("New follower journal last index", 3, newFollowerActorContext.getReplicatedLog().lastIndex()); - verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), LEADER_ID, FOLLOWER_ID, NEW_SERVER_ID); + verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(FOLLOWER_ID), votingServer(NEW_SERVER_ID)); // Verify new server config was applied in both followers @@ -172,6 +178,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 @@ -207,13 +228,15 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { 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(), LEADER_ID, NEW_SERVER_ID); + 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(), LEADER_ID, NEW_SERVER_ID); + verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), votingServer(LEADER_ID), + votingServer(NEW_SERVER_ID)); // Verify new server config was applied in the new follower @@ -246,13 +269,15 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { 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(), LEADER_ID, NEW_SERVER_ID); + 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(), LEADER_ID, NEW_SERVER_ID); + verifyServerConfigurationPayloadEntry(newFollowerActorContext.getReplicatedLog(), votingServer(LEADER_ID), + nonVotingServer(NEW_SERVER_ID)); // Verify new server config was applied in the new follower @@ -276,7 +301,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { assertEquals("Leader commit index", 1, leaderActorContext.getCommitIndex()); assertEquals("Leader last applied index", 1, leaderActorContext.getLastApplied()); verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), - LEADER_ID, NEW_SERVER_ID, NEW_SERVER_ID2); + votingServer(LEADER_ID), nonVotingServer(NEW_SERVER_ID), nonVotingServer(NEW_SERVER_ID2)); } @Test @@ -307,12 +332,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()); @@ -325,7 +353,7 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { assertEquals("Leader commit index", 1, leaderActorContext.getCommitIndex()); assertEquals("Leader last applied index", 1, leaderActorContext.getLastApplied()); verifyServerConfigurationPayloadEntry(leaderActorContext.getReplicatedLog(), - LEADER_ID, NEW_SERVER_ID, NEW_SERVER_ID2); + votingServer(LEADER_ID), votingServer(NEW_SERVER_ID), nonVotingServer(NEW_SERVER_ID2)); // Verify ServerConfigurationPayload entry in the new follower @@ -336,9 +364,185 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { } @Test - public void testAddServerWithInstallSnapshotTimeout() throws Exception { - newFollowerRaftActor.underlyingActor().setDropMessageOfType(InstallSnapshot.SERIALIZABLE_CLASS); + public void testAddServerWithPriorSnapshotInProgress() throws Exception { + RaftActorContext initialActorContext = new MockRaftActorContext(); + initialActorContext.setCommitIndex(-1); + initialActorContext.setLastApplied(-1); + initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().build()); + + TestActorRef leaderActor = actorFactory.createTestActor( + MockLeaderRaftActor.props(ImmutableMap.of(), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); + RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + + TestActorRef leaderCollectorActor = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID + "Collector")); + leaderRaftActor.setCollectorActor(leaderCollectorActor); + + // 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 + + 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(), + initialActorContext).withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID)); + + MockLeaderRaftActor leaderRaftActor = leaderActor.underlyingActor(); + RaftActorContext leaderActorContext = leaderRaftActor.getRaftActorContext(); + + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(1); + + TestActorRef leaderCollectorActor = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID + "Collector")); + leaderRaftActor.setCollectorActor(leaderCollectorActor); + + // Drop commit message so the snapshot doesn't complete. + leaderRaftActor.setDropMessageOfType(String.class); + + 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(); + initialActorContext.setCommitIndex(-1); + initialActorContext.setLastApplied(-1); + initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().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(); + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(100); + + TestActorRef leaderCollectorActor = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID + "Collector")); + leaderRaftActor.setCollectorActor(leaderCollectorActor); + + // 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(); + initialActorContext.setCommitIndex(-1); + initialActorContext.setLastApplied(-1); + initialActorContext.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().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(); + + ((DefaultConfigParamsImpl)leaderActorContext.getConfigParams()).setElectionTimeoutFactor(8); + + TestActorRef leaderCollectorActor = actorFactory.createTestActor( + MessageCollectorActor.props().withDispatcher(Dispatchers.DefaultDispatcherId()), + actorFactory.generateActorId(LEADER_ID + "Collector")); + leaderRaftActor.setCollectorActor(leaderCollectorActor); + + // 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(); initialActorContext.setCommitIndex(-1); initialActorContext.setLastApplied(-1); @@ -353,8 +557,13 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { 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()); @@ -401,14 +610,22 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { expectFirstMatching(leaderActor, AddServer.class); } - private void verifyServerConfigurationPayloadEntry(ReplicatedLog log, String... cNew) { + private ServerInfo votingServer(String id) { + return new ServerInfo(id, true); + } + + private ServerInfo nonVotingServer(String id) { + return new ServerInfo(id, false); + } + + 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(cNew), Sets.newHashSet(payload.getNewServerConfig())); + assertEquals("getNewServerConfig", Sets.newHashSet(expected), Sets.newHashSet(payload.getServerConfig())); } - private RaftActorContext newFollowerContext(String id, TestActorRef actor) { + private static RaftActorContext newFollowerContext(String id, TestActorRef actor) { DefaultConfigParamsImpl configParams = new DefaultConfigParamsImpl(); configParams.setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS)); configParams.setElectionTimeoutFactor(100000); @@ -424,10 +641,41 @@ public class RaftActorServerConfigurationSupportTest extends AbstractActorTest { return followerActorContext; } - public static class MockLeaderRaftActor extends MockRaftActor { + 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 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++) { @@ -465,26 +713,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) {