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%2FRaftActorTest.java;h=87e40f236cf5c6dfee6e03ea711d61d342d9b345;hb=03e752cbd625921ece92c5281cd4e1a8c81b3210;hp=69af77c7496b26ed02033d97fd8572025814cc9f;hpb=a564badf85498c6f2c85c831dcb6319953214b4f;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java index 69af77c749..a0663efdd5 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/RaftActorTest.java @@ -1,57 +1,93 @@ package org.opendaylight.controller.cluster.raft; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.anyObject; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import akka.actor.ActorRef; import akka.actor.ActorSystem; import akka.actor.PoisonPill; import akka.actor.Props; import akka.actor.Terminated; -import akka.event.Logging; import akka.japi.Creator; +import akka.japi.Procedure; +import akka.pattern.Patterns; import akka.persistence.RecoveryCompleted; +import akka.persistence.SaveSnapshotFailure; import akka.persistence.SaveSnapshotSuccess; import akka.persistence.SnapshotMetadata; import akka.persistence.SnapshotOffer; +import akka.persistence.SnapshotSelectionCriteria; import akka.testkit.JavaTestKit; import akka.testkit.TestActorRef; +import akka.util.Timeout; import com.google.common.base.Optional; import com.google.common.collect.Lists; +import com.google.common.util.concurrent.Uninterruptibles; import com.google.protobuf.ByteString; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; import org.junit.After; +import org.junit.Assert; +import org.junit.Before; import org.junit.Test; import org.opendaylight.controller.cluster.DataPersistenceProvider; import org.opendaylight.controller.cluster.datastore.DataPersistenceProviderMonitor; +import org.opendaylight.controller.cluster.notifications.RoleChanged; import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries; +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.CaptureSnapshot; import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply; +import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat; +import org.opendaylight.controller.cluster.raft.behaviors.Follower; +import org.opendaylight.controller.cluster.raft.behaviors.Leader; import org.opendaylight.controller.cluster.raft.client.messages.FindLeader; import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply; +import org.opendaylight.controller.cluster.raft.messages.AppendEntries; +import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload; +import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor; import org.opendaylight.controller.cluster.raft.utils.MockAkkaJournal; import org.opendaylight.controller.cluster.raft.utils.MockSnapshotStore; +import scala.concurrent.Await; +import scala.concurrent.Future; +import scala.concurrent.duration.Duration; import scala.concurrent.duration.FiniteDuration; -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.IOException; -import java.io.ObjectInputStream; -import java.io.ObjectOutputStream; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotEquals; -import static org.mockito.Mockito.mock; - public class RaftActorTest extends AbstractActorTest { + private TestActorFactory factory; + + @Before + public void setUp(){ + factory = new TestActorFactory(getSystem()); + } @After - public void tearDown() { + public void tearDown() throws Exception { + factory.close(); MockAkkaJournal.clearJournal(); MockSnapshotStore.setMockSnapshot(null); } @@ -59,36 +95,44 @@ public class RaftActorTest extends AbstractActorTest { public static class MockRaftActor extends RaftActor { private final DataPersistenceProvider dataPersistenceProvider; + private final RaftActor delegate; + private final CountDownLatch recoveryComplete = new CountDownLatch(1); + private final List state; + private ActorRef roleChangeNotifier; + private final CountDownLatch initializeBehaviorComplete = new CountDownLatch(1); public static final class MockRaftActorCreator implements Creator { + private static final long serialVersionUID = 1L; private final Map peerAddresses; private final String id; private final Optional config; private final DataPersistenceProvider dataPersistenceProvider; + private final ActorRef roleChangeNotifier; private MockRaftActorCreator(Map peerAddresses, String id, - Optional config, DataPersistenceProvider dataPersistenceProvider) { + Optional config, DataPersistenceProvider dataPersistenceProvider, + ActorRef roleChangeNotifier) { this.peerAddresses = peerAddresses; this.id = id; this.config = config; this.dataPersistenceProvider = dataPersistenceProvider; + this.roleChangeNotifier = roleChangeNotifier; } @Override public MockRaftActor create() throws Exception { - return new MockRaftActor(id, peerAddresses, config, dataPersistenceProvider); + MockRaftActor mockRaftActor = new MockRaftActor(id, peerAddresses, config, + dataPersistenceProvider); + mockRaftActor.roleChangeNotifier = this.roleChangeNotifier; + return mockRaftActor; } } - private final CountDownLatch recoveryComplete = new CountDownLatch(1); - private final CountDownLatch applyRecoverySnapshot = new CountDownLatch(1); - private final CountDownLatch applyStateLatch = new CountDownLatch(1); - - private final List state; - - public MockRaftActor(String id, Map peerAddresses, Optional config, DataPersistenceProvider dataPersistenceProvider) { + public MockRaftActor(String id, Map peerAddresses, Optional config, + DataPersistenceProvider dataPersistenceProvider) { super(id, peerAddresses, config); state = new ArrayList<>(); + this.delegate = mock(RaftActor.class); if(dataPersistenceProvider == null){ this.dataPersistenceProvider = new PersistentDataProvider(); } else { @@ -104,8 +148,12 @@ public class RaftActorTest extends AbstractActorTest { } } - public CountDownLatch getApplyRecoverySnapshotLatch(){ - return applyRecoverySnapshot; + public void waitForInitializeBehaviorComplete() { + try { + assertEquals("Behavior initialized", true, initializeBehaviorComplete.await(5, TimeUnit.SECONDS)); + } catch (InterruptedException e) { + e.printStackTrace(); + } } public List getState() { @@ -114,17 +162,22 @@ public class RaftActorTest extends AbstractActorTest { public static Props props(final String id, final Map peerAddresses, Optional config){ - return Props.create(new MockRaftActorCreator(peerAddresses, id, config, null)); + return Props.create(new MockRaftActorCreator(peerAddresses, id, config, null, null)); } public static Props props(final String id, final Map peerAddresses, Optional config, DataPersistenceProvider dataPersistenceProvider){ - return Props.create(new MockRaftActorCreator(peerAddresses, id, config, dataPersistenceProvider)); + return Props.create(new MockRaftActorCreator(peerAddresses, id, config, dataPersistenceProvider, null)); } + public static Props props(final String id, final Map peerAddresses, + Optional config, ActorRef roleChangeNotifier){ + return Props.create(new MockRaftActorCreator(peerAddresses, id, config, null, roleChangeNotifier)); + } @Override protected void applyState(ActorRef clientActor, String identifier, Object data) { - applyStateLatch.countDown(); + delegate.applyState(clientActor, identifier, data); + LOG.info("applyState called"); } @Override @@ -142,17 +195,23 @@ public class RaftActorTest extends AbstractActorTest { @Override protected void onRecoveryComplete() { + delegate.onRecoveryComplete(); recoveryComplete.countDown(); } @Override - protected void applyRecoverySnapshot(ByteString snapshot) { - applyRecoverySnapshot.countDown(); + protected void initializeBehavior() { + super.initializeBehavior(); + initializeBehaviorComplete.countDown(); + } + + @Override + protected void applyRecoverySnapshot(byte[] bytes) { + delegate.applyRecoverySnapshot(bytes); try { - Object data = toObject(snapshot); - System.out.println("!!!!!applyRecoverySnapshot: "+data); + Object data = toObject(bytes); if (data instanceof List) { - state.addAll((List) data); + state.addAll((List) data); } } catch (Exception e) { e.printStackTrace(); @@ -160,12 +219,15 @@ public class RaftActorTest extends AbstractActorTest { } @Override protected void createSnapshot() { + delegate.createSnapshot(); } - @Override protected void applySnapshot(ByteString snapshot) { + @Override protected void applySnapshot(byte [] snapshot) { + delegate.applySnapshot(snapshot); } @Override protected void onStateChanged() { + delegate.onStateChanged(); } @Override @@ -173,16 +235,21 @@ public class RaftActorTest extends AbstractActorTest { return this.dataPersistenceProvider; } + @Override + protected Optional getRoleChangeNotifier() { + return Optional.fromNullable(roleChangeNotifier); + } + @Override public String persistenceId() { return this.getId(); } - private Object toObject(ByteString bs) throws ClassNotFoundException, IOException { + private Object toObject(byte[] bs) throws ClassNotFoundException, IOException { Object obj = null; ByteArrayInputStream bis = null; ObjectInputStream ois = null; try { - bis = new ByteArrayInputStream(bs.toByteArray()); + bis = new ByteArrayInputStream(bs); ois = new ObjectInputStream(bis); obj = ois.readObject(); } finally { @@ -210,57 +277,75 @@ public class RaftActorTest extends AbstractActorTest { super(actorSystem); raftActor = this.getSystem().actorOf(MockRaftActor.props(actorName, - Collections.EMPTY_MAP, Optional.absent()), actorName); + Collections.emptyMap(), Optional.absent()), actorName); } - public boolean waitForStartup(){ + public ActorRef getRaftActor() { + return raftActor; + } + + public boolean waitForLogMessage(final Class logEventClass, String message){ // Wait for a specific log message to show up return - new JavaTestKit.EventFilter(Logging.Info.class + new JavaTestKit.EventFilter(logEventClass ) { @Override protected Boolean run() { return true; } }.from(raftActor.path().toString()) - .message("Switching from behavior Candidate to Leader") + .message(message) .occurrences(1).exec(); } - public void findLeader(final String expectedLeader){ - raftActor.tell(new FindLeader(), getRef()); - - FindLeaderReply reply = expectMsgClass(duration("5 seconds"), FindLeaderReply.class); - assertEquals("getLeaderActor", expectedLeader, reply.getLeaderActor()); + protected void waitUntilLeader(){ + waitUntilLeader(raftActor); } - public ActorRef getRaftActor() { - return raftActor; + protected void waitUntilLeader(ActorRef actorRef) { + FiniteDuration duration = Duration.create(100, TimeUnit.MILLISECONDS); + for(int i = 0; i < 20 * 5; i++) { + Future future = Patterns.ask(actorRef, new FindLeader(), new Timeout(duration)); + try { + FindLeaderReply resp = (FindLeaderReply) Await.result(future, duration); + if(resp.getLeaderActor() != null) { + return; + } + } catch(TimeoutException e) { + } catch(Exception e) { + System.err.println("FindLeader threw ex"); + e.printStackTrace(); + } + + + Uninterruptibles.sleepUninterruptibly(50, TimeUnit.MILLISECONDS); + } + + Assert.fail("Leader not found for actorRef " + actorRef.path()); } + } @Test public void testConstruction() { - boolean started = new RaftActorTestKit(getSystem(), "testConstruction").waitForStartup(); - assertEquals(true, started); + new RaftActorTestKit(getSystem(), "testConstruction").waitUntilLeader(); } @Test public void testFindLeaderWhenLeaderIsSelf(){ RaftActorTestKit kit = new RaftActorTestKit(getSystem(), "testFindLeader"); - kit.waitForStartup(); - kit.findLeader(kit.getRaftActor().path().toString()); + kit.waitUntilLeader(); } @Test public void testRaftActorRecovery() throws Exception { new JavaTestKit(getSystem()) {{ - String persistenceId = "follower10"; + String persistenceId = factory.generateActorId("follower-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); // Set the heartbeat interval high to essentially disable election otherwise the test @@ -268,8 +353,8 @@ public class RaftActorTest extends AbstractActorTest { // log entry. config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - ActorRef followerActor = getSystem().actorOf(MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config)), persistenceId); + ActorRef followerActor = factory.createActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config)), persistenceId); watch(followerActor); @@ -281,15 +366,15 @@ public class RaftActorTest extends AbstractActorTest { int lastAppliedDuringSnapshotCapture = 3; int lastIndexDuringSnapshotCapture = 4; - // 4 messages as part of snapshot, which are applied to state - ByteString snapshotBytes = fromObject(Arrays.asList( - new MockRaftActorContext.MockPayload("A"), - new MockRaftActorContext.MockPayload("B"), - new MockRaftActorContext.MockPayload("C"), - new MockRaftActorContext.MockPayload("D"))); + // 4 messages as part of snapshot, which are applied to state + ByteString snapshotBytes = fromObject(Arrays.asList( + new MockRaftActorContext.MockPayload("A"), + new MockRaftActorContext.MockPayload("B"), + new MockRaftActorContext.MockPayload("C"), + new MockRaftActorContext.MockPayload("D"))); Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(), - snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1 , + snapshotUnappliedEntries, lastIndexDuringSnapshotCapture, 1, lastAppliedDuringSnapshotCapture, 1); MockSnapshotStore.setMockSnapshot(snapshot); MockSnapshotStore.setPersistenceId(persistenceId); @@ -322,8 +407,8 @@ public class RaftActorTest extends AbstractActorTest { unwatch(followerActor); //reinstate the actor - TestActorRef ref = TestActorRef.create(getSystem(), - MockRaftActor.props(persistenceId, Collections.EMPTY_MAP, + TestActorRef ref = factory.createTestActor( + MockRaftActor.props(persistenceId, Collections.emptyMap(), Optional.of(config))); ref.underlyingActor().waitForRecoveryComplete(); @@ -349,31 +434,32 @@ public class RaftActorTest extends AbstractActorTest { public void testHandleRecoveryWhenDataPersistenceRecoveryApplicable() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testHandleRecoveryWhenDataPersistenceRecoveryApplicable"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config)), persistenceId); + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config)), persistenceId); + + MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); - ByteString snapshotBytes = fromObject(Arrays.asList( + // Wait for akka's recovery to complete so it doesn't interfere. + mockRaftActor.waitForRecoveryComplete(); + + ByteString snapshotBytes = fromObject(Arrays.asList( new MockRaftActorContext.MockPayload("A"), new MockRaftActorContext.MockPayload("B"), new MockRaftActorContext.MockPayload("C"), new MockRaftActorContext.MockPayload("D"))); Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(), - Lists.newArrayList(), 3, 1 ,3, 1); - - MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); + Lists.newArrayList(), 3, 1, 3, 1); mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot)); - CountDownLatch applyRecoverySnapshotLatch = mockRaftActor.getApplyRecoverySnapshotLatch(); - - assertEquals("apply recovery snapshot", true, applyRecoverySnapshotLatch.await(5, TimeUnit.SECONDS)); + verify(mockRaftActor.delegate).applyRecoverySnapshot(eq(snapshotBytes.toByteArray())); mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A"))); @@ -402,10 +488,6 @@ public class RaftActorTest extends AbstractActorTest { mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class)); - mockRaftActor.waitForRecoveryComplete(); - - mockActorRef.tell(PoisonPill.getInstance(), getRef()); - }}; } @@ -419,31 +501,32 @@ public class RaftActorTest extends AbstractActorTest { public void testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testHandleRecoveryWhenDataPersistenceRecoveryNotApplicable"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config), new DataPersistenceProviderMonitor()), persistenceId); + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), new DataPersistenceProviderMonitor()), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); - ByteString snapshotBytes = fromObject(Arrays.asList( + // Wait for akka's recovery to complete so it doesn't interfere. + mockRaftActor.waitForRecoveryComplete(); + + ByteString snapshotBytes = fromObject(Arrays.asList( new MockRaftActorContext.MockPayload("A"), new MockRaftActorContext.MockPayload("B"), new MockRaftActorContext.MockPayload("C"), new MockRaftActorContext.MockPayload("D"))); Snapshot snapshot = Snapshot.create(snapshotBytes.toByteArray(), - Lists.newArrayList(), 3, 1 ,3, 1); + Lists.newArrayList(), 3, 1, 3, 1); mockRaftActor.onReceiveRecover(new SnapshotOffer(new SnapshotMetadata(persistenceId, 100, 100), snapshot)); - CountDownLatch applyRecoverySnapshotLatch = mockRaftActor.getApplyRecoverySnapshotLatch(); - - assertEquals("apply recovery snapshot", false, applyRecoverySnapshotLatch.await(1, TimeUnit.SECONDS)); + verify(mockRaftActor.delegate, times(0)).applyRecoverySnapshot(any(byte[].class)); mockRaftActor.onReceiveRecover(new ReplicatedLogImplEntry(0, 1, new MockRaftActorContext.MockPayload("A"))); @@ -470,9 +553,6 @@ public class RaftActorTest extends AbstractActorTest { mockRaftActor.onReceiveRecover(mock(RecoveryCompleted.class)); - mockRaftActor.waitForRecoveryComplete(); - - mockActorRef.tell(PoisonPill.getInstance(), getRef()); }}; } @@ -481,7 +561,7 @@ public class RaftActorTest extends AbstractActorTest { public void testUpdatingElectionTermCallsDataPersistence() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testUpdatingElectionTermCallsDataPersistence"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); @@ -491,8 +571,8 @@ public class RaftActorTest extends AbstractActorTest { DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); dataPersistenceProviderMonitor.setPersistLatch(persistLatch); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProviderMonitor), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); @@ -500,9 +580,8 @@ public class RaftActorTest extends AbstractActorTest { assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS)); - mockActorRef.tell(PoisonPill.getInstance(), getRef()); - } + }; } @@ -510,28 +589,27 @@ public class RaftActorTest extends AbstractActorTest { public void testAddingReplicatedLogEntryCallsDataPersistence() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testAddingReplicatedLogEntryCallsDataPersistence"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - CountDownLatch persistLatch = new CountDownLatch(1); - DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); - dataPersistenceProviderMonitor.setPersistLatch(persistLatch); + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProvider), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); - mockRaftActor.getRaftActorContext().getReplicatedLog().appendAndPersist(new MockRaftActorContext.MockReplicatedLogEntry(10, 10, mock(Payload.class))); + MockRaftActorContext.MockReplicatedLogEntry logEntry = new MockRaftActorContext.MockReplicatedLogEntry(10, 10, mock(Payload.class)); - assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS)); + mockRaftActor.getRaftActorContext().getReplicatedLog().appendAndPersist(logEntry); - mockActorRef.tell(PoisonPill.getInstance(), getRef()); + verify(dataPersistenceProvider).persist(eq(logEntry), any(Procedure.class)); } + }; } @@ -539,18 +617,16 @@ public class RaftActorTest extends AbstractActorTest { public void testRemovingReplicatedLogEntryCallsDataPersistence() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testRemovingReplicatedLogEntryCallsDataPersistence"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - CountDownLatch persistLatch = new CountDownLatch(2); - DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); - dataPersistenceProviderMonitor.setPersistLatch(persistLatch); + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProvider), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); @@ -558,11 +634,10 @@ public class RaftActorTest extends AbstractActorTest { mockRaftActor.getRaftActorContext().getReplicatedLog().removeFromAndPersist(0); - assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS)); - - mockActorRef.tell(PoisonPill.getInstance(), getRef()); + verify(dataPersistenceProvider, times(2)).persist(anyObject(), any(Procedure.class)); } + }; } @@ -570,28 +645,25 @@ public class RaftActorTest extends AbstractActorTest { public void testApplyLogEntriesCallsDataPersistence() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testApplyLogEntriesCallsDataPersistence"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - CountDownLatch persistLatch = new CountDownLatch(1); - DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); - dataPersistenceProviderMonitor.setPersistLatch(persistLatch); + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProvider), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); mockRaftActor.onReceiveCommand(new ApplyLogEntries(10)); - assertEquals("Persist called", true, persistLatch.await(5, TimeUnit.SECONDS)); - - mockActorRef.tell(PoisonPill.getInstance(), getRef()); + verify(dataPersistenceProvider, times(1)).persist(anyObject(), any(Procedure.class)); } + }; } @@ -599,34 +671,35 @@ public class RaftActorTest extends AbstractActorTest { public void testCaptureSnapshotReplyCallsDataPersistence() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testCaptureSnapshotReplyCallsDataPersistence"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - CountDownLatch persistLatch = new CountDownLatch(1); - DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); - dataPersistenceProviderMonitor.setSaveSnapshotLatch(persistLatch); + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + TestActorRef mockActorRef = factory.createTestActor( + MockRaftActor.props(persistenceId, Collections.emptyMap(), + Optional.of(config), dataPersistenceProvider), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); - ByteString snapshotBytes = fromObject(Arrays.asList( + ByteString snapshotBytes = fromObject(Arrays.asList( new MockRaftActorContext.MockPayload("A"), new MockRaftActorContext.MockPayload("B"), new MockRaftActorContext.MockPayload("C"), new MockRaftActorContext.MockPayload("D"))); - mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1,1,-1,1)); + mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1)); - mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes)); + RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext(); - assertEquals("Save Snapshot called", true, persistLatch.await(5, TimeUnit.SECONDS)); + mockRaftActor.setCurrentBehavior(new Leader(raftActorContext)); - mockActorRef.tell(PoisonPill.getInstance(), getRef()); + mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray())); + + verify(dataPersistenceProvider).saveSnapshot(anyObject()); } }; @@ -636,45 +709,488 @@ public class RaftActorTest extends AbstractActorTest { public void testSaveSnapshotSuccessCallsDataPersistence() throws Exception { new JavaTestKit(getSystem()) { { - String persistenceId = "testSaveSnapshotSuccessCallsDataPersistence"; + String persistenceId = factory.generateActorId("leader-"); DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); - CountDownLatch deleteMessagesLatch = new CountDownLatch(1); - CountDownLatch deleteSnapshotsLatch = new CountDownLatch(1); - DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); - dataPersistenceProviderMonitor.setDeleteMessagesLatch(deleteMessagesLatch); - dataPersistenceProviderMonitor.setDeleteSnapshotsLatch(deleteSnapshotsLatch); + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); - TestActorRef mockActorRef = TestActorRef.create(getSystem(), MockRaftActor.props(persistenceId, - Collections.EMPTY_MAP, Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProvider), persistenceId); MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); - ByteString snapshotBytes = fromObject(Arrays.asList( + mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class))); + mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 1, mock(Payload.class))); + mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 2, mock(Payload.class))); + mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 3, mock(Payload.class))); + mockRaftActor.getReplicatedLog().append(new MockRaftActorContext.MockReplicatedLogEntry(1, 4, mock(Payload.class))); + + ByteString snapshotBytes = fromObject(Arrays.asList( new MockRaftActorContext.MockPayload("A"), new MockRaftActorContext.MockPayload("B"), new MockRaftActorContext.MockPayload("C"), new MockRaftActorContext.MockPayload("D"))); - mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1,1,-1,1)); + RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext(); + mockRaftActor.setCurrentBehavior(new Follower(raftActorContext)); + + mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, 2, 1)); - mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes)); + verify(mockRaftActor.delegate).createSnapshot(); + + mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray())); mockRaftActor.onReceiveCommand(new SaveSnapshotSuccess(new SnapshotMetadata("foo", 100, 100))); - assertEquals("Delete Messages called", true, deleteMessagesLatch.await(5, TimeUnit.SECONDS)); + verify(dataPersistenceProvider).deleteSnapshots(any(SnapshotSelectionCriteria.class)); + + verify(dataPersistenceProvider).deleteMessages(100); + + assertEquals(2, mockRaftActor.getReplicatedLog().size()); + + assertNotNull(mockRaftActor.getReplicatedLog().get(3)); + assertNotNull(mockRaftActor.getReplicatedLog().get(4)); + + // Index 2 will not be in the log because it was removed due to snapshotting + assertNull(mockRaftActor.getReplicatedLog().get(2)); + + } + }; + } + + @Test + public void testApplyState() throws Exception { + + new JavaTestKit(getSystem()) { + { + String persistenceId = factory.generateActorId("leader-"); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + + config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); + + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProvider), persistenceId); + + MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); + + ReplicatedLogEntry entry = new MockRaftActorContext.MockReplicatedLogEntry(1, 5, + new MockRaftActorContext.MockPayload("F")); - assertEquals("Delete Snapshots called", true, deleteSnapshotsLatch.await(5, TimeUnit.SECONDS)); + mockRaftActor.onReceiveCommand(new ApplyState(mockActorRef, "apply-state", entry)); - mockActorRef.tell(PoisonPill.getInstance(), getRef()); + verify(mockRaftActor.delegate).applyState(eq(mockActorRef), eq("apply-state"), anyObject()); } }; } + @Test + public void testApplySnapshot() throws Exception { + new JavaTestKit(getSystem()) { + { + String persistenceId = factory.generateActorId("leader-"); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + + config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); + + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + + MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); + + ReplicatedLog oldReplicatedLog = mockRaftActor.getReplicatedLog(); + + oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1, 0, mock(Payload.class))); + oldReplicatedLog.append(new MockRaftActorContext.MockReplicatedLogEntry(1, 1, mock(Payload.class))); + oldReplicatedLog.append( + new MockRaftActorContext.MockReplicatedLogEntry(1, 2, + mock(Payload.class))); + + ByteString snapshotBytes = fromObject(Arrays.asList( + new MockRaftActorContext.MockPayload("A"), + new MockRaftActorContext.MockPayload("B"), + new MockRaftActorContext.MockPayload("C"), + new MockRaftActorContext.MockPayload("D"))); + + Snapshot snapshot = mock(Snapshot.class); + + doReturn(snapshotBytes.toByteArray()).when(snapshot).getState(); + + doReturn(3L).when(snapshot).getLastAppliedIndex(); + + mockRaftActor.onReceiveCommand(new ApplySnapshot(snapshot)); + + verify(mockRaftActor.delegate).applySnapshot(eq(snapshot.getState())); + + assertTrue("The replicatedLog should have changed", + oldReplicatedLog != mockRaftActor.getReplicatedLog()); + + assertEquals("lastApplied should be same as in the snapshot", + (Long) 3L, mockRaftActor.getLastApplied()); + + assertEquals(0, mockRaftActor.getReplicatedLog().size()); + + } + }; + } + + @Test + public void testSaveSnapshotFailure() throws Exception { + new JavaTestKit(getSystem()) { + { + String persistenceId = factory.generateActorId("leader-"); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + + config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + DataPersistenceProviderMonitor dataPersistenceProviderMonitor = new DataPersistenceProviderMonitor(); + + TestActorRef mockActorRef = factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), dataPersistenceProviderMonitor), persistenceId); + + MockRaftActor mockRaftActor = mockActorRef.underlyingActor(); + + ByteString snapshotBytes = fromObject(Arrays.asList( + new MockRaftActorContext.MockPayload("A"), + new MockRaftActorContext.MockPayload("B"), + new MockRaftActorContext.MockPayload("C"), + new MockRaftActorContext.MockPayload("D"))); + + RaftActorContext raftActorContext = mockRaftActor.getRaftActorContext(); + + mockRaftActor.setCurrentBehavior(new Leader(raftActorContext)); + + mockRaftActor.onReceiveCommand(new CaptureSnapshot(-1, 1, -1, 1)); + + mockRaftActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray())); + + mockRaftActor.onReceiveCommand(new SaveSnapshotFailure(new SnapshotMetadata("foobar", 10L, 1234L), + new Exception())); + + assertEquals("Snapshot index should not have advanced because save snapshot failed", -1, + mockRaftActor.getReplicatedLog().getSnapshotIndex()); + + } + }; + } + + @Test + public void testRaftRoleChangeNotifier() throws Exception { + new JavaTestKit(getSystem()) {{ + ActorRef notifierActor = factory.createActor(Props.create(MessageCollectorActor.class)); + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + String persistenceId = factory.generateActorId("notifier-"); + + factory.createTestActor(MockRaftActor.props(persistenceId, + Collections.emptyMap(), Optional.of(config), notifierActor), persistenceId); + + // sleeping for a minimum of 2 seconds, if it spans more its fine. + Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); + + List matches = MessageCollectorActor.getAllMatching(notifierActor, RoleChanged.class); + assertNotNull(matches); + assertEquals(3, matches.size()); + + // check if the notifier got a role change from null to Follower + RoleChanged raftRoleChanged = matches.get(0); + assertEquals(persistenceId, raftRoleChanged.getMemberId()); + assertNull(raftRoleChanged.getOldRole()); + assertEquals(RaftState.Follower.name(), raftRoleChanged.getNewRole()); + + // check if the notifier got a role change from Follower to Candidate + raftRoleChanged = matches.get(1); + assertEquals(persistenceId, raftRoleChanged.getMemberId()); + assertEquals(RaftState.Follower.name(), raftRoleChanged.getOldRole()); + assertEquals(RaftState.Candidate.name(), raftRoleChanged.getNewRole()); + + // check if the notifier got a role change from Candidate to Leader + raftRoleChanged = matches.get(2); + assertEquals(persistenceId, raftRoleChanged.getMemberId()); + assertEquals(RaftState.Candidate.name(), raftRoleChanged.getOldRole()); + assertEquals(RaftState.Leader.name(), raftRoleChanged.getNewRole()); + }}; + } + + @Test + public void testFakeSnapshotsForLeaderWithInRealSnapshots() throws Exception { + new JavaTestKit(getSystem()) { + { + String persistenceId = factory.generateActorId("leader-"); + String follower1Id = factory.generateActorId("follower-"); + + ActorRef followerActor1 = + factory.createActor(Props.create(MessageCollectorActor.class)); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); + + Map peerAddresses = new HashMap<>(); + peerAddresses.put(follower1Id, followerActor1.path().toString()); + + TestActorRef mockActorRef = TestActorRef.create(getSystem(), + MockRaftActor.props(persistenceId, peerAddresses, + Optional.of(config), dataPersistenceProvider), persistenceId); + + MockRaftActor leaderActor = mockActorRef.underlyingActor(); + leaderActor.getRaftActorContext().setCommitIndex(4); + leaderActor.getRaftActorContext().setLastApplied(4); + leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId); + + leaderActor.waitForInitializeBehaviorComplete(); + + // create 8 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot + + Leader leader = new Leader(leaderActor.getRaftActorContext()); + leaderActor.setCurrentBehavior(leader); + assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state()); + + MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder(); + leaderActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(0, 8, 1).build()); + + assertEquals(8, leaderActor.getReplicatedLog().size()); + + leaderActor.onReceiveCommand(new CaptureSnapshot(6, 1, 4, 1)); + leaderActor.getRaftActorContext().setSnapshotCaptureInitiated(true); + verify(leaderActor.delegate).createSnapshot(); + + assertEquals(8, leaderActor.getReplicatedLog().size()); + + assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state()); + //fake snapshot on index 5 + leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 5, 1)); + + assertEquals(8, leaderActor.getReplicatedLog().size()); + + //fake snapshot on index 6 + assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state()); + leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 6, 1)); + assertEquals(8, leaderActor.getReplicatedLog().size()); + + assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state()); + + assertEquals(8, leaderActor.getReplicatedLog().size()); + + ByteString snapshotBytes = fromObject(Arrays.asList( + new MockRaftActorContext.MockPayload("foo-0"), + new MockRaftActorContext.MockPayload("foo-1"), + new MockRaftActorContext.MockPayload("foo-2"), + new MockRaftActorContext.MockPayload("foo-3"), + new MockRaftActorContext.MockPayload("foo-4"))); + leaderActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray())); + assertFalse(leaderActor.getRaftActorContext().isSnapshotCaptureInitiated()); + + // capture snapshot reply should remove the snapshotted entries only + assertEquals(3, leaderActor.getReplicatedLog().size()); + assertEquals(7, leaderActor.getReplicatedLog().lastIndex()); + + // add another non-replicated entry + leaderActor.getReplicatedLog().append( + new ReplicatedLogImplEntry(8, 1, new MockRaftActorContext.MockPayload("foo-8"))); + + //fake snapshot on index 7, since lastApplied = 7 , we would keep the last applied + leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 7, 1)); + assertEquals(2, leaderActor.getReplicatedLog().size()); + assertEquals(8, leaderActor.getReplicatedLog().lastIndex()); + + } + }; + } + + @Test + public void testFakeSnapshotsForFollowerWithInRealSnapshots() throws Exception { + new JavaTestKit(getSystem()) { + { + String persistenceId = factory.generateActorId("follower-"); + String leaderId = factory.generateActorId("leader-"); + + + ActorRef leaderActor1 = + factory.createActor(Props.create(MessageCollectorActor.class)); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); + + Map peerAddresses = new HashMap<>(); + peerAddresses.put(leaderId, leaderActor1.path().toString()); + + TestActorRef mockActorRef = TestActorRef.create(getSystem(), + MockRaftActor.props(persistenceId, peerAddresses, + Optional.of(config), dataPersistenceProvider), persistenceId); + + MockRaftActor followerActor = mockActorRef.underlyingActor(); + followerActor.getRaftActorContext().setCommitIndex(4); + followerActor.getRaftActorContext().setLastApplied(4); + followerActor.getRaftActorContext().getTermInformation().update(1, persistenceId); + + followerActor.waitForInitializeBehaviorComplete(); + + // create 8 entries in the log - 0 to 4 are applied and will get picked up as part of the capture snapshot + Follower follower = new Follower(followerActor.getRaftActorContext()); + followerActor.setCurrentBehavior(follower); + assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state()); + + MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder(); + followerActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(0, 6, 1).build()); + + // log as indices 0-5 + assertEquals(6, followerActor.getReplicatedLog().size()); + + //snapshot on 4 + followerActor.onReceiveCommand(new CaptureSnapshot(5, 1, 4, 1)); + followerActor.getRaftActorContext().setSnapshotCaptureInitiated(true); + verify(followerActor.delegate).createSnapshot(); + + assertEquals(6, followerActor.getReplicatedLog().size()); + + //fake snapshot on index 6 + List entries = + Arrays.asList( + (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 6, + new MockRaftActorContext.MockPayload("foo-6")) + ); + followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 5, 1, entries, 5, 5)); + assertEquals(7, followerActor.getReplicatedLog().size()); + + //fake snapshot on index 7 + assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state()); + + entries = + Arrays.asList( + (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 7, + new MockRaftActorContext.MockPayload("foo-7")) + ); + followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 6, 1, entries, 6, 6)); + assertEquals(8, followerActor.getReplicatedLog().size()); + + assertEquals(RaftState.Follower, followerActor.getCurrentBehavior().state()); + + + ByteString snapshotBytes = fromObject(Arrays.asList( + new MockRaftActorContext.MockPayload("foo-0"), + new MockRaftActorContext.MockPayload("foo-1"), + new MockRaftActorContext.MockPayload("foo-2"), + new MockRaftActorContext.MockPayload("foo-3"), + new MockRaftActorContext.MockPayload("foo-4"))); + followerActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray())); + assertFalse(followerActor.getRaftActorContext().isSnapshotCaptureInitiated()); + + // capture snapshot reply should remove the snapshotted entries only + assertEquals(3, followerActor.getReplicatedLog().size()); //indexes 5,6,7 left in the log + assertEquals(7, followerActor.getReplicatedLog().lastIndex()); + + entries = + Arrays.asList( + (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(1, 8, + new MockRaftActorContext.MockPayload("foo-7")) + ); + // send an additional entry 8 with leaderCommit = 7 + followerActor.onReceiveCommand(new AppendEntries(1, leaderId, 7, 1, entries, 7, 7)); + + // 7 and 8, as lastapplied is 7 + assertEquals(2, followerActor.getReplicatedLog().size()); + + } + }; + } + + @Test + public void testFakeSnapshotsForLeaderWithInInitiateSnapshots() throws Exception { + new JavaTestKit(getSystem()) { + { + String persistenceId = factory.generateActorId("leader-"); + String follower1Id = factory.generateActorId("follower-"); + String follower2Id = factory.generateActorId("follower-"); + + ActorRef followerActor1 = + factory.createActor(Props.create(MessageCollectorActor.class), follower1Id); + ActorRef followerActor2 = + factory.createActor(Props.create(MessageCollectorActor.class), follower2Id); + + DefaultConfigParamsImpl config = new DefaultConfigParamsImpl(); + config.setHeartBeatInterval(new FiniteDuration(1, TimeUnit.DAYS)); + config.setIsolatedLeaderCheckInterval(new FiniteDuration(1, TimeUnit.DAYS)); + + DataPersistenceProvider dataPersistenceProvider = mock(DataPersistenceProvider.class); + + Map peerAddresses = new HashMap<>(); + peerAddresses.put(follower1Id, followerActor1.path().toString()); + peerAddresses.put(follower2Id, followerActor2.path().toString()); + + TestActorRef mockActorRef = factory.createTestActor( + MockRaftActor.props(persistenceId, peerAddresses, + Optional.of(config), dataPersistenceProvider), persistenceId); + + MockRaftActor leaderActor = mockActorRef.underlyingActor(); + leaderActor.getRaftActorContext().setCommitIndex(9); + leaderActor.getRaftActorContext().setLastApplied(9); + leaderActor.getRaftActorContext().getTermInformation().update(1, persistenceId); + + leaderActor.waitForInitializeBehaviorComplete(); + + Leader leader = new Leader(leaderActor.getRaftActorContext()); + leaderActor.setCurrentBehavior(leader); + assertEquals(RaftState.Leader, leaderActor.getCurrentBehavior().state()); + + // create 5 entries in the log + MockRaftActorContext.MockReplicatedLogBuilder logBuilder = new MockRaftActorContext.MockReplicatedLogBuilder(); + leaderActor.getRaftActorContext().setReplicatedLog(logBuilder.createEntries(5, 10, 1).build()); + //set the snapshot index to 4 , 0 to 4 are snapshotted + leaderActor.getRaftActorContext().getReplicatedLog().setSnapshotIndex(4); + assertEquals(5, leaderActor.getReplicatedLog().size()); + + leaderActor.onReceiveCommand(new AppendEntriesReply(follower1Id, 1, true, 9, 1)); + assertEquals(5, leaderActor.getReplicatedLog().size()); + + // set the 2nd follower nextIndex to 1 which has been snapshotted + leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 0, 1)); + assertEquals(5, leaderActor.getReplicatedLog().size()); + + // simulate a real snapshot + leaderActor.onReceiveCommand(new SendHeartBeat()); + assertEquals(5, leaderActor.getReplicatedLog().size()); + assertEquals(String.format("expected to be Leader but was %s. Current Leader = %s ", + leaderActor.getCurrentBehavior().state(), leaderActor.getLeaderId()) + , RaftState.Leader, leaderActor.getCurrentBehavior().state()); + + + //reply from a slow follower does not initiate a fake snapshot + leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 9, 1)); + assertEquals("Fake snapshot should not happen when Initiate is in progress", 5, leaderActor.getReplicatedLog().size()); + + ByteString snapshotBytes = fromObject(Arrays.asList( + new MockRaftActorContext.MockPayload("foo-0"), + new MockRaftActorContext.MockPayload("foo-1"), + new MockRaftActorContext.MockPayload("foo-2"), + new MockRaftActorContext.MockPayload("foo-3"), + new MockRaftActorContext.MockPayload("foo-4"))); + leaderActor.onReceiveCommand(new CaptureSnapshotReply(snapshotBytes.toByteArray())); + assertFalse(leaderActor.getRaftActorContext().isSnapshotCaptureInitiated()); + + assertEquals("Real snapshot didn't clear the log till lastApplied", 0, leaderActor.getReplicatedLog().size()); + + //reply from a slow follower after should not raise errors + leaderActor.onReceiveCommand(new AppendEntriesReply(follower2Id, 1, true, 5, 1)); + assertEquals(0, leaderActor.getReplicatedLog().size()); + } + }; + } + private ByteString fromObject(Object snapshot) throws Exception { ByteArrayOutputStream b = null; ObjectOutputStream o = null; @@ -694,4 +1210,5 @@ public class RaftActorTest extends AbstractActorTest { } } } + }