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%2Fbehaviors%2FAbstractRaftActorBehaviorTest.java;h=7695d05133e6cd18e4d97a35282d9931995a2f01;hb=HEAD;hp=3cd373adf4319e36f5101c418e330adc90c7931d;hpb=a0c5aba42aa36337ff1c6760175918b786897c9e;p=controller.git diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java index 3cd373adf4..3497840b38 100644 --- a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java +++ b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/AbstractRaftActorBehaviorTest.java @@ -1,572 +1,338 @@ +/* + * Copyright (c) 2014, 2015 Cisco Systems, Inc. and others. All rights reserved. + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v1.0 which accompanies this distribution, + * and is available at http://www.eclipse.org/legal/epl-v10.html + */ + package org.opendaylight.controller.cluster.raft.behaviors; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; + import akka.actor.ActorRef; -import akka.actor.Props; -import akka.testkit.JavaTestKit; +import akka.protobuf.ByteString; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.junit.After; +import org.junit.Assert; import org.junit.Test; import org.opendaylight.controller.cluster.raft.AbstractActorTest; import org.opendaylight.controller.cluster.raft.MockRaftActorContext; import org.opendaylight.controller.cluster.raft.RaftActorContext; import org.opendaylight.controller.cluster.raft.RaftState; import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.TestActorFactory; import org.opendaylight.controller.cluster.raft.messages.AppendEntries; import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply; +import org.opendaylight.controller.cluster.raft.messages.Payload; import org.opendaylight.controller.cluster.raft.messages.RaftRPC; import org.opendaylight.controller.cluster.raft.messages.RequestVote; import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply; -import org.opendaylight.controller.cluster.raft.utils.DoNothingActor; +import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry; +import org.opendaylight.controller.cluster.raft.policy.RaftPolicy; +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.LoggerFactory; -import java.util.ArrayList; -import java.util.List; +public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest { -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; + protected final TestActorFactory actorFactory = new TestActorFactory(getSystem()); + + private final ActorRef behaviorActor = actorFactory.createActor( + MessageCollectorActor.props(), actorFactory.generateActorId("behavior")); -public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest { + RaftActorBehavior behavior; - private final ActorRef behaviorActor = getSystem().actorOf(Props.create( - DoNothingActor.class)); + @After + public void tearDown() { + if (behavior != null) { + behavior.close(); + } + + actorFactory.close(); + + InMemoryJournal.clear(); + InMemorySnapshotStore.clear(); + } /** * This test checks that when a new Raft RPC message is received with a newer * term the RaftActor gets into the Follower state. - * - * @throws Exception */ @Test - public void testHandleRaftRPCWithNewerTerm() throws Exception { - new JavaTestKit(getSystem()) {{ + public void testHandleRaftRPCWithNewerTerm() { + MockRaftActorContext actorContext = createActorContext(); - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createAppendEntriesWithNewerTerm()); - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createAppendEntriesReplyWithNewerTerm()); - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createRequestVoteWithNewerTerm()); - assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(getTestActor(), + assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, behaviorActor, createRequestVoteReplyWithNewerTerm()); - - - }}; } - /** - * This test verifies that when an AppendEntries RPC is received by a RaftActor - * with a commitIndex that is greater than what has been applied to the - * state machine of the RaftActor, the RaftActor applies the state and - * sets it current applied state to the commitIndex of the sender. - * - * @throws Exception - */ - @Test - public void testHandleAppendEntriesWithNewerCommitIndex() throws Exception { - new JavaTestKit(getSystem()) {{ - - RaftActorContext context = - createActorContext(); - - context.setLastApplied(100); - setLastLogEntry((MockRaftActorContext) context, 0, 0, ""); - - // The new commitIndex is 101 - AppendEntries appendEntries = - new AppendEntries(100, "leader-1", 0, 0, null, 101); - - RaftState raftState = - createBehavior(context).handleMessage(getRef(), appendEntries); - - assertEquals(101L, context.getLastApplied()); - - }}; - } /** * This test verifies that when an AppendEntries is received with a term that * is less that the currentTerm of the RaftActor then the RaftActor does not - * change it's state and it responds back with a failure - * - * @throws Exception + * change it's state and it responds back with a failure. */ @Test - public void testHandleAppendEntriesSenderTermLessThanReceiverTerm() - throws Exception { - new JavaTestKit(getSystem()) {{ - - MockRaftActorContext context = (MockRaftActorContext) - createActorContext(); + public void testHandleAppendEntriesSenderTermLessThanReceiverTerm() { + MockRaftActorContext context = createActorContext(); + short payloadVersion = 5; + context.setPayloadVersion(payloadVersion); - // First set the receivers term to a high number (1000) - context.getTermInformation().update(1000, "test"); + // First set the receivers term to a high number (1000) + context.getTermInformation().update(1000, "test"); - AppendEntries appendEntries = - new AppendEntries(100, "leader-1", 0, 0, null, 101); + AppendEntries appendEntries = new AppendEntries(100, "leader-1", 0, 0, Collections.emptyList(), 101, -1, + (short)4); - RaftActorBehavior behavior = createBehavior(context); + behavior = createBehavior(context); - // Send an unknown message so that the state of the RaftActor remains unchanged - RaftState expected = behavior.handleMessage(getRef(), "unknown"); + RaftState expected = behavior.state(); - RaftState raftState = - behavior.handleMessage(getRef(), appendEntries); + RaftActorBehavior raftBehavior = behavior.handleMessage(behaviorActor, appendEntries); - assertEquals(expected, raftState); + assertEquals("Raft state", expected, raftBehavior.state()); - // Also expect an AppendEntriesReply to be sent where success is false - final Boolean out = new ExpectMsg(duration("1 seconds"), - "AppendEntriesReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof AppendEntriesReply) { - AppendEntriesReply reply = (AppendEntriesReply) in; - return reply.isSuccess(); - } else { - throw noMatch(); - } - } - }.get(); + // Also expect an AppendEntriesReply to be sent where success is false - assertEquals(false, out); + AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching( + behaviorActor, AppendEntriesReply.class); - - }}; + assertEquals("isSuccess", false, reply.isSuccess()); + assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion()); } - /** - * This test verifies that when an AppendEntries is received a specific prevLogTerm - * which does not match the term that is in RaftActors log entry at prevLogIndex - * then the RaftActor does not change it's state and it returns a failure. - * - * @throws Exception - */ + @Test - public void testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() - throws Exception { - new JavaTestKit(getSystem()) {{ + public void testHandleAppendEntriesAddSameEntryToLog() { + MockRaftActorContext context = createActorContext(); - MockRaftActorContext context = (MockRaftActorContext) - createActorContext(); + context.getTermInformation().update(2, "test"); - // First set the receivers term to lower number - context.getTermInformation().update(95, "test"); + // Prepare the receivers log + MockRaftActorContext.MockPayload payload = new MockRaftActorContext.MockPayload("zero"); + setLastLogEntry(context, 2, 0, payload); - // Set the last log entry term for the receiver to be greater than - // what we will be sending as the prevLogTerm in AppendEntries - MockRaftActorContext.MockReplicatedLog mockReplicatedLog = - setLastLogEntry(context, 20, 0, ""); + List entries = new ArrayList<>(); + entries.add(new SimpleReplicatedLogEntry(0, 2, payload)); - // Also set the entry at index 0 with term 20 which will be greater - // than the prevLogTerm sent by the sender - mockReplicatedLog.setReplicatedLogEntry( - new MockRaftActorContext.MockReplicatedLogEntry(20, 0, "")); + final AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 2, -1, (short)0); - // AppendEntries is now sent with a bigger term - // this will set the receivers term to be the same as the sender's term - AppendEntries appendEntries = - new AppendEntries(100, "leader-1", 0, 0, null, 101); + behavior = createBehavior(context); - RaftActorBehavior behavior = createBehavior(context); + assertFalse("This test should be overridden when testing Candidate", behavior instanceof Candidate); - // Send an unknown message so that the state of the RaftActor remains unchanged - RaftState expected = behavior.handleMessage(getRef(), "unknown"); + RaftState expected = behavior.state(); - RaftState raftState = - behavior.handleMessage(getRef(), appendEntries); + // Check that the behavior does not handle unknwon message + assertNull(behavior.handleMessage(behaviorActor, "unknown")); - assertEquals(expected, raftState); + RaftActorBehavior raftBehavior = behavior.handleMessage(behaviorActor, appendEntries); - // Also expect an AppendEntriesReply to be sent where success is false - final Boolean out = new ExpectMsg(duration("1 seconds"), - "AppendEntriesReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof AppendEntriesReply) { - AppendEntriesReply reply = (AppendEntriesReply) in; - return reply.isSuccess(); - } else { - throw noMatch(); - } - } - }.get(); + assertEquals("Raft state", expected, raftBehavior.state()); - assertEquals(false, out); + assertEquals("ReplicatedLog size", 1, context.getReplicatedLog().size()); + handleAppendEntriesAddSameEntryToLogReply(behaviorActor); + } - }}; + protected void handleAppendEntriesAddSameEntryToLogReply(final ActorRef replyActor) { + AppendEntriesReply reply = MessageCollectorActor.getFirstMatching(replyActor, AppendEntriesReply.class); + Assert.assertNull("Expected no AppendEntriesReply", reply); } /** - * This test verifies that when a new AppendEntries message is received with - * new entries and the logs of the sender and receiver match that the new - * entries get added to the log and the log is incremented by the number of - * entries received in appendEntries - * - * @throws Exception + * This test verifies that when a RequestVote is received by the RaftActor + * with the senders' log is more up to date than the receiver that the receiver grants + * the vote to the sender. */ @Test - public void testHandleAppendEntriesAddNewEntries() throws Exception { - new JavaTestKit(getSystem()) {{ - - MockRaftActorContext context = (MockRaftActorContext) - createActorContext(); - - // First set the receivers term to lower number - context.getTermInformation().update(1, "test"); - - // Prepare the receivers log - MockRaftActorContext.SimpleReplicatedLog log = - new MockRaftActorContext.SimpleReplicatedLog(); - log.append( - new MockRaftActorContext.MockReplicatedLogEntry(1, 0, "zero")); - log.append( - new MockRaftActorContext.MockReplicatedLogEntry(1, 1, "one")); - log.append( - new MockRaftActorContext.MockReplicatedLogEntry(1, 2, "two")); - - context.setReplicatedLog(log); - - // Prepare the entries to be sent with AppendEntries - List entries = new ArrayList<>(); - entries.add( - new MockRaftActorContext.MockReplicatedLogEntry(1, 3, "three")); - entries.add( - new MockRaftActorContext.MockReplicatedLogEntry(1, 4, "four")); - - // Send appendEntries with the same term as was set on the receiver - // before the new behavior was created (1 in this case) - // This will not work for a Candidate because as soon as a Candidate - // is created it increments the term - AppendEntries appendEntries = - new AppendEntries(1, "leader-1", 2, 1, entries, 101); - - RaftActorBehavior behavior = createBehavior(context); - - if (AbstractRaftActorBehaviorTest.this instanceof CandidateTest) { - // Resetting the Candidates term to make sure it will match - // the term sent by AppendEntries. If this was not done then - // the test will fail because the Candidate will assume that - // the message was sent to it from a lower term peer and will - // thus respond with a failure - context.getTermInformation().update(1, "test"); - } - - // Send an unknown message so that the state of the RaftActor remains unchanged - RaftState expected = behavior.handleMessage(getRef(), "unknown"); - - RaftState raftState = - behavior.handleMessage(getRef(), appendEntries); - - assertEquals(expected, raftState); - assertEquals(5, log.last().getIndex() + 1); - assertNotNull(log.get(3)); - assertNotNull(log.get(4)); + public void testHandleRequestVoteWhenSenderLogMoreUpToDate() { + MockRaftActorContext context = createActorContext(); - // Also expect an AppendEntriesReply to be sent where success is false - final Boolean out = new ExpectMsg(duration("1 seconds"), - "AppendEntriesReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof AppendEntriesReply) { - AppendEntriesReply reply = (AppendEntriesReply) in; - return reply.isSuccess(); - } else { - throw noMatch(); - } - } - }.get(); + behavior = createBehavior(context); - assertEquals(true, out); + context.getTermInformation().update(1, "test"); + behavior.handleMessage(behaviorActor, new RequestVote(context.getTermInformation().getCurrentTerm(), + "test", 10000, 999)); - }}; + RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor, + RequestVoteReply.class); + assertEquals("isVoteGranted", true, reply.isVoteGranted()); } /** - * This test verifies that when a new AppendEntries message is received with - * new entries and the logs of the sender and receiver are out-of-sync that - * the log is first corrected by removing the out of sync entries from the - * log and then adding in the new entries sent with the AppendEntries message - * - * @throws Exception + * This test verifies that when a RaftActor receives a RequestVote message + * with a term that is greater than it's currentTerm but a less up-to-date + * log then the receiving RaftActor will not grant the vote to the sender. */ @Test - public void testHandleAppendEntriesCorrectReceiverLogEntries() - throws Exception { - new JavaTestKit(getSystem()) {{ - - MockRaftActorContext context = (MockRaftActorContext) - createActorContext(); - - // First set the receivers term to lower number - context.getTermInformation().update(2, "test"); - - // Prepare the receivers log - MockRaftActorContext.SimpleReplicatedLog log = - new MockRaftActorContext.SimpleReplicatedLog(); - log.append( - new MockRaftActorContext.MockReplicatedLogEntry(1, 0, "zero")); - log.append( - new MockRaftActorContext.MockReplicatedLogEntry(1, 1, "one")); - log.append( - new MockRaftActorContext.MockReplicatedLogEntry(1, 2, "two")); - - context.setReplicatedLog(log); - - // Prepare the entries to be sent with AppendEntries - List entries = new ArrayList<>(); - entries.add( - new MockRaftActorContext.MockReplicatedLogEntry(2, 2, "two-1")); - entries.add( - new MockRaftActorContext.MockReplicatedLogEntry(2, 3, "three")); - - // Send appendEntries with the same term as was set on the receiver - // before the new behavior was created (1 in this case) - // This will not work for a Candidate because as soon as a Candidate - // is created it increments the term - AppendEntries appendEntries = - new AppendEntries(2, "leader-1", 1, 1, entries, 101); - - RaftActorBehavior behavior = createBehavior(context); - - if (AbstractRaftActorBehaviorTest.this instanceof CandidateTest) { - // Resetting the Candidates term to make sure it will match - // the term sent by AppendEntries. If this was not done then - // the test will fail because the Candidate will assume that - // the message was sent to it from a lower term peer and will - // thus respond with a failure - context.getTermInformation().update(2, "test"); - } - - // Send an unknown message so that the state of the RaftActor remains unchanged - RaftState expected = behavior.handleMessage(getRef(), "unknown"); - - RaftState raftState = - behavior.handleMessage(getRef(), appendEntries); + public void testHandleRequestVoteWhenSenderLogLessUptoDate() { + MockRaftActorContext context = createActorContext(); - assertEquals(expected, raftState); + behavior = createBehavior(context); - // The entry at index 2 will be found out-of-sync with the leader - // and will be removed - // Then the two new entries will be added to the log - // Thus making the log to have 4 entries - assertEquals(4, log.last().getIndex() + 1); - assertNotNull(log.get(2)); + context.getTermInformation().update(1, "test"); - // Check that the entry at index 2 has the new data - assertEquals("two-1", log.get(2).getData()); - assertNotNull(log.get(3)); + int index = 2000; + setLastLogEntry(context, context.getTermInformation().getCurrentTerm(), index, + new MockRaftActorContext.MockPayload("")); - // Also expect an AppendEntriesReply to be sent where success is false - final Boolean out = new ExpectMsg(duration("1 seconds"), - "AppendEntriesReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof AppendEntriesReply) { - AppendEntriesReply reply = (AppendEntriesReply) in; - return reply.isSuccess(); - } else { - throw noMatch(); - } - } - }.get(); + behavior.handleMessage(behaviorActor, new RequestVote( + context.getTermInformation().getCurrentTerm(), "test", + index - 1, context.getTermInformation().getCurrentTerm())); - assertEquals(true, out); + RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor, + RequestVoteReply.class); + assertEquals("isVoteGranted", false, reply.isVoteGranted()); + } - }}; - } /** - * This test verifies that when a RequestVote is received by the RaftActor - * with a term which is greater than the RaftActors' currentTerm and the - * senders' log is more upto date than the receiver that the receiver grants - * the vote to the sender + * This test verifies that the receiving RaftActor will not grant a vote + * to a sender if the sender's term is lesser than the currentTerm of the + * recipient RaftActor. */ @Test - public void testHandleRequestVoteWhenSenderTermGreaterThanCurrentTermAndSenderLogMoreUpToDate() { - new JavaTestKit(getSystem()) {{ - - new Within(duration("1 seconds")) { - protected void run() { - - RaftActorBehavior follower = createBehavior( - createActorContext(behaviorActor)); - - follower.handleMessage(getTestActor(), - new RequestVote(1000, "test", 10000, 999)); - - final Boolean out = - new ExpectMsg(duration("1 seconds"), - "RequestVoteReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof RequestVoteReply) { - RequestVoteReply reply = - (RequestVoteReply) in; - return reply.isVoteGranted(); - } else { - throw noMatch(); - } - } - }.get(); - - assertEquals(true, out); - } - }; - }}; - } + public void testHandleRequestVoteWhenSenderTermLessThanCurrentTerm() { + MockRaftActorContext context = createActorContext(); - /** - * This test verifies that when a RaftActor receives a RequestVote message - * with a term that is greater than it's currentTerm but a less up-to-date - * log then the receiving RaftActor will not grant the vote to the sender - */ - @Test - public void testHandleRequestVoteWhenSenderTermGreaterThanCurrentTermButSenderLogLessUptoDate() { - new JavaTestKit(getSystem()) {{ - - new Within(duration("1 seconds")) { - protected void run() { - - RaftActorContext actorContext = - createActorContext(behaviorActor); - - MockRaftActorContext.MockReplicatedLog - log = new MockRaftActorContext.MockReplicatedLog(); - log.setReplicatedLogEntry( - new MockRaftActorContext.MockReplicatedLogEntry(20000, - 1000000, "")); - log.setLast( - new MockRaftActorContext.MockReplicatedLogEntry(20000, - 1000000, "") - ); - - ((MockRaftActorContext) actorContext).setReplicatedLog(log); - - RaftActorBehavior follower = createBehavior(actorContext); - - follower.handleMessage(getTestActor(), - new RequestVote(1000, "test", 10000, 999)); - - final Boolean out = - new ExpectMsg(duration("1 seconds"), - "RequestVoteReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof RequestVoteReply) { - RequestVoteReply reply = - (RequestVoteReply) in; - return reply.isVoteGranted(); - } else { - throw noMatch(); - } - } - }.get(); - - assertEquals(false, out); - } - }; - }}; - } + context.getTermInformation().update(1000, null); + behavior = createBehavior(context); + behavior.handleMessage(behaviorActor, new RequestVote(999, "test", 10000, 999)); + + RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(behaviorActor, + RequestVoteReply.class); + assertEquals("isVoteGranted", false, reply.isVoteGranted()); + } - /** - * This test verifies that the receiving RaftActor will not grant a vote - * to a sender if the sender's term is lesser than the currentTerm of the - * recipient RaftActor - */ @Test - public void testHandleRequestVoteWhenSenderTermLessThanCurrentTerm() { - new JavaTestKit(getSystem()) {{ - - new Within(duration("1 seconds")) { - protected void run() { - - RaftActorContext context = - createActorContext(behaviorActor); - - context.getTermInformation().update(1000, null); - - RaftActorBehavior follower = createBehavior(context); - - follower.handleMessage(getTestActor(), - new RequestVote(999, "test", 10000, 999)); - - final Boolean out = - new ExpectMsg(duration("1 seconds"), - "RequestVoteReply") { - // do not put code outside this method, will run afterwards - protected Boolean match(Object in) { - if (in instanceof RequestVoteReply) { - RequestVoteReply reply = - (RequestVoteReply) in; - return reply.isVoteGranted(); - } else { - throw noMatch(); - } - } - }.get(); - - assertEquals(false, out); - } - }; - }}; + public void testPerformSnapshot() { + MockRaftActorContext context = new MockRaftActorContext("test", getSystem(), behaviorActor); + AbstractRaftActorBehavior abstractBehavior = (AbstractRaftActorBehavior) createBehavior(context); + if (abstractBehavior instanceof Candidate) { + return; + } + + context.getTermInformation().update(1, "test"); + + //log has 1 entry with replicatedToAllIndex = 0, does not do anything, returns the + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 1, 1).build()); + context.setLastApplied(0); + abstractBehavior.performSnapshotWithoutCapture(0); + assertEquals(-1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(1, context.getReplicatedLog().size()); + + //2 entries, lastApplied still 0, no purging. + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build()); + context.setLastApplied(0); + abstractBehavior.performSnapshotWithoutCapture(0); + assertEquals(-1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(2, context.getReplicatedLog().size()); + + //2 entries, lastApplied still 0, no purging. + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 2, 1).build()); + context.setLastApplied(1); + abstractBehavior.performSnapshotWithoutCapture(0); + assertEquals(0, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(1, context.getReplicatedLog().size()); + + // 5 entries, lastApplied =2 and replicatedIndex = 3, but since we want to keep the lastapplied, indices 0 and + // 1 will only get purged + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 5, 1).build()); + context.setLastApplied(2); + abstractBehavior.performSnapshotWithoutCapture(3); + assertEquals(1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(3, context.getReplicatedLog().size()); + + // scenario where Last applied > Replicated to all index (becoz of a slow follower) + context.setReplicatedLog(new MockRaftActorContext.MockReplicatedLogBuilder().createEntries(0, 3, 1).build()); + context.setLastApplied(2); + abstractBehavior.performSnapshotWithoutCapture(1); + assertEquals(1, abstractBehavior.getReplicatedToAllIndex()); + assertEquals(1, context.getReplicatedLog().size()); } - protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm( - ActorRef actorRef, RaftRPC rpc) { - RaftActorContext actorContext = createActorContext(); - setLastLogEntry( - (MockRaftActorContext) actorContext, 0, 0, ""); + protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(final MockRaftActorContext actorContext, + final ActorRef actorRef, final RaftRPC rpc) { + + Payload payload = new MockRaftActorContext.MockPayload(""); + setLastLogEntry(actorContext, 1, 0, payload); + actorContext.getTermInformation().update(1, "test"); + + RaftActorBehavior origBehavior = createBehavior(actorContext); + RaftActorBehavior raftBehavior = origBehavior.handleMessage(actorRef, rpc); - RaftState raftState = createBehavior(actorContext) - .handleMessage(actorRef, rpc); + assertEquals("New raft state", RaftState.Follower, raftBehavior.state()); + assertEquals("New election term", rpc.getTerm(), actorContext.getTermInformation().getCurrentTerm()); - assertEquals(RaftState.Follower, raftState); + origBehavior.close(); + raftBehavior.close(); } - protected MockRaftActorContext.MockReplicatedLog setLastLogEntry( - MockRaftActorContext actorContext, long term, long index, Object data) { - return setLastLogEntry(actorContext, - new MockRaftActorContext.MockReplicatedLogEntry(term, index, data)); + protected MockRaftActorContext.SimpleReplicatedLog setLastLogEntry( + final MockRaftActorContext actorContext, final long term, final long index, final Payload data) { + return setLastLogEntry(actorContext, new SimpleReplicatedLogEntry(index, term, data)); } - protected MockRaftActorContext.MockReplicatedLog setLastLogEntry( - MockRaftActorContext actorContext, ReplicatedLogEntry logEntry) { - MockRaftActorContext.MockReplicatedLog - log = new MockRaftActorContext.MockReplicatedLog(); - // By default MockReplicateLog has last entry set to (1,1,"") - log.setLast(logEntry); + protected MockRaftActorContext.SimpleReplicatedLog setLastLogEntry(final MockRaftActorContext actorContext, + final ReplicatedLogEntry logEntry) { + MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog(); + log.append(logEntry); actorContext.setReplicatedLog(log); return log; } - protected abstract RaftActorBehavior createBehavior( - RaftActorContext actorContext); + protected abstract T createBehavior(RaftActorContext actorContext); + + protected final T createBehavior(final MockRaftActorContext actorContext) { + T ret = createBehavior((RaftActorContext)actorContext); + actorContext.setCurrentBehavior(ret); + return ret; + } protected RaftActorBehavior createBehavior() { return createBehavior(createActorContext()); } - protected RaftActorContext createActorContext() { + protected MockRaftActorContext createActorContext() { return new MockRaftActorContext(); } - protected RaftActorContext createActorContext(ActorRef actor) { + protected MockRaftActorContext createActorContext(final ActorRef actor) { return new MockRaftActorContext("test", getSystem(), actor); } protected AppendEntries createAppendEntriesWithNewerTerm() { - return new AppendEntries(100, "leader-1", 0, 0, null, 1); + return new AppendEntries(100, "leader-1", 0, 0, Collections.emptyList(), 1, -1, (short)0); } protected AppendEntriesReply createAppendEntriesReplyWithNewerTerm() { - return new AppendEntriesReply(100, false); + return new AppendEntriesReply("follower-1", 100, false, 100, 100, (short)0); } protected RequestVote createRequestVoteWithNewerTerm() { @@ -577,6 +343,32 @@ public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest { return new RequestVoteReply(100, false); } + protected ByteString toByteString(final Map state) { + ByteArrayOutputStream bos = new ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(bos)) { + oos.writeObject(state); + return ByteString.copyFrom(bos.toByteArray()); + } catch (IOException e) { + throw new AssertionError("IOException occurred converting Map to Bytestring", e); + } + } + + protected void logStart(final String name) { + LoggerFactory.getLogger(getClass()).info("Starting " + name); + } + protected RaftPolicy createRaftPolicy(final boolean automaticElectionsEnabled, + final boolean applyModificationToStateBeforeConsensus) { + return new RaftPolicy() { + @Override + public boolean automaticElectionsEnabled() { + return automaticElectionsEnabled; + } + @Override + public boolean applyModificationToStateBeforeConsensus() { + return applyModificationToStateBeforeConsensus; + } + }; + } }