+/*
+ * 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.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
import akka.actor.ActorRef;
import akka.actor.Props;
+import akka.dispatch.Dispatchers;
import akka.testkit.JavaTestKit;
+import akka.testkit.TestActorRef;
+import com.google.common.base.Optional;
+import com.google.common.base.Stopwatch;
+import com.google.common.base.Throwables;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.io.ByteSource;
+import com.google.common.util.concurrent.Uninterruptibles;
import com.google.protobuf.ByteString;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import org.junit.After;
+import org.junit.Assert;
import org.junit.Test;
import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
+import org.opendaylight.controller.cluster.raft.MockRaftActor;
+import org.opendaylight.controller.cluster.raft.MockRaftActor.Builder;
+import org.opendaylight.controller.cluster.raft.MockRaftActor.MockSnapshotState;
import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
+import org.opendaylight.controller.cluster.raft.RaftActorSnapshotCohort;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
+import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
+import org.opendaylight.controller.cluster.raft.base.messages.FollowerInitialSyncUpStatus;
+import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
+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.ApplyJournalEntries;
+import org.opendaylight.controller.cluster.raft.persisted.ByteState;
+import org.opendaylight.controller.cluster.raft.persisted.ServerConfigurationPayload;
+import org.opendaylight.controller.cluster.raft.persisted.ServerInfo;
+import org.opendaylight.controller.cluster.raft.persisted.SimpleReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.persisted.Snapshot;
+import org.opendaylight.controller.cluster.raft.persisted.Snapshot.State;
+import org.opendaylight.controller.cluster.raft.persisted.UpdateElectionTerm;
+import org.opendaylight.controller.cluster.raft.policy.DisableElectionsRaftPolicy;
+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 scala.concurrent.duration.FiniteDuration;
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.ObjectOutputStream;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+public class FollowerTest extends AbstractRaftActorBehaviorTest<Follower> {
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
+ private final TestActorRef<MessageCollectorActor> followerActor = actorFactory.createTestActor(
+ Props.create(MessageCollectorActor.class), actorFactory.generateActorId("follower"));
-public class FollowerTest extends AbstractRaftActorBehaviorTest {
+ private final TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
+ Props.create(MessageCollectorActor.class), actorFactory.generateActorId("leader"));
- private final ActorRef followerActor = getSystem().actorOf(Props.create(
- DoNothingActor.class));
+ private Follower follower;
+ private final short payloadVersion = 5;
- @Override protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
- return new Follower(actorContext);
+ @Override
+ @After
+ public void tearDown() throws Exception {
+ if (follower != null) {
+ follower.close();
+ }
+
+ super.tearDown();
}
- @Override protected RaftActorContext createActorContext() {
+ @Override
+ protected Follower createBehavior(RaftActorContext actorContext) {
+ return spy(new Follower(actorContext));
+ }
+
+ @Override
+ protected MockRaftActorContext createActorContext() {
return createActorContext(followerActor);
}
- protected RaftActorContext createActorContext(ActorRef actorRef){
- return new MockRaftActorContext("test", getSystem(), actorRef);
+ @Override
+ protected MockRaftActorContext createActorContext(ActorRef actorRef) {
+ MockRaftActorContext context = new MockRaftActorContext("follower", getSystem(), actorRef);
+ context.setPayloadVersion(payloadVersion);
+ return context;
+ }
+
+ @Test
+ public void testThatAnElectionTimeoutIsTriggered() {
+ MockRaftActorContext actorContext = createActorContext();
+ follower = new Follower(actorContext);
+
+ MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class,
+ actorContext.getConfigParams().getElectionTimeOutInterval().$times(6).toMillis());
}
@Test
- public void testThatAnElectionTimeoutIsTriggered(){
- new JavaTestKit(getSystem()) {{
+ public void testHandleElectionTimeoutWhenNoLeaderMessageReceived() {
+ logStart("testHandleElectionTimeoutWhenNoLeaderMessageReceived");
- new Within(DefaultConfigParamsImpl.HEART_BEAT_INTERVAL.$times(6)) {
- protected void run() {
+ MockRaftActorContext context = createActorContext();
+ follower = new Follower(context);
- Follower follower = new Follower(createActorContext(getTestActor()));
+ Uninterruptibles.sleepUninterruptibly(context.getConfigParams().getElectionTimeOutInterval().toMillis(),
+ TimeUnit.MILLISECONDS);
+ RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
- final Boolean out = new ExpectMsg<Boolean>(DefaultConfigParamsImpl.HEART_BEAT_INTERVAL.$times(6), "ElectionTimeout") {
- // do not put code outside this method, will run afterwards
- protected Boolean match(Object in) {
- if (in instanceof ElectionTimeout) {
- return true;
- } else {
- throw noMatch();
- }
- }
- }.get();
+ assertTrue(raftBehavior instanceof Candidate);
+ }
- assertEquals(true, out);
- }
- };
- }};
+ @Test
+ public void testHandleElectionTimeoutWhenLeaderMessageReceived() {
+ logStart("testHandleElectionTimeoutWhenLeaderMessageReceived");
+
+ MockRaftActorContext context = createActorContext();
+ ((DefaultConfigParamsImpl) context.getConfigParams())
+ .setHeartBeatInterval(new FiniteDuration(100, TimeUnit.MILLISECONDS));
+ ((DefaultConfigParamsImpl) context.getConfigParams()).setElectionTimeoutFactor(4);
+
+ follower = new Follower(context);
+ context.setCurrentBehavior(follower);
+
+ Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
+ .getElectionTimeOutInterval().toMillis() - 100, TimeUnit.MILLISECONDS);
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
+ -1, -1, (short) 1));
+
+ Uninterruptibles.sleepUninterruptibly(130, TimeUnit.MILLISECONDS);
+ RaftActorBehavior raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
+ assertTrue(raftBehavior instanceof Follower);
+
+ Uninterruptibles.sleepUninterruptibly(context.getConfigParams()
+ .getElectionTimeOutInterval().toMillis() - 150, TimeUnit.MILLISECONDS);
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", -1, -1, Collections.emptyList(),
+ -1, -1, (short) 1));
+
+ Uninterruptibles.sleepUninterruptibly(200, TimeUnit.MILLISECONDS);
+ raftBehavior = follower.handleMessage(leaderActor, ElectionTimeout.INSTANCE);
+ assertTrue(raftBehavior instanceof Follower);
}
@Test
- public void testHandleElectionTimeout(){
- RaftActorContext raftActorContext = createActorContext();
- Follower follower =
- new Follower(raftActorContext);
+ public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull() {
+ logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull");
- RaftActorBehavior raftBehavior =
- follower.handleMessage(followerActor, new ElectionTimeout());
+ MockRaftActorContext context = createActorContext();
+ long term = 1000;
+ context.getTermInformation().update(term, null);
- assertTrue(raftBehavior instanceof Candidate);
+ follower = createBehavior(context);
+
+ follower.handleMessage(leaderActor, new RequestVote(term, "test", 10000, 999));
+
+ RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
+
+ assertEquals("isVoteGranted", true, reply.isVoteGranted());
+ assertEquals("getTerm", term, reply.getTerm());
+ verify(follower).scheduleElection(any(FiniteDuration.class));
}
@Test
- public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull(){
- new JavaTestKit(getSystem()) {{
+ public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId() {
+ logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId");
- new Within(duration("1 seconds")) {
- protected void run() {
+ MockRaftActorContext context = createActorContext();
+ long term = 1000;
+ context.getTermInformation().update(term, "test");
- RaftActorContext context = createActorContext(getTestActor());
+ follower = createBehavior(context);
- context.getTermInformation().update(1000, null);
+ follower.handleMessage(leaderActor, new RequestVote(term, "candidate", 10000, 999));
- RaftActorBehavior follower = createBehavior(context);
+ RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
- follower.handleMessage(getTestActor(), new RequestVote(1000, "test", 10000, 999));
+ assertEquals("isVoteGranted", false, reply.isVoteGranted());
+ verify(follower, never()).scheduleElection(any(FiniteDuration.class));
+ }
- final Boolean out = new ExpectMsg<Boolean>(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);
- }
- };
- }};
+ @Test
+ public void testHandleFirstAppendEntries() throws Exception {
+ logStart("testHandleFirstAppendEntries");
+
+ MockRaftActorContext context = createActorContext();
+ context.getReplicatedLog().clear(0,2);
+ context.getReplicatedLog().append(newReplicatedLogEntry(1,100, "bar"));
+ context.getReplicatedLog().setSnapshotIndex(99);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ Assert.assertEquals(1, context.getReplicatedLog().size());
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
+
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+ assertTrue("append entries reply should be true", reply.isSuccess());
}
@Test
- public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId(){
- new JavaTestKit(getSystem()) {{
+ public void testHandleFirstAppendEntriesWithPrevIndexMinusOne() throws Exception {
+ logStart("testHandleFirstAppendEntries");
- new Within(duration("1 seconds")) {
- protected void run() {
+ MockRaftActorContext context = createActorContext();
- RaftActorContext context = createActorContext(getTestActor());
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
- context.getTermInformation().update(1000, "test");
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
- RaftActorBehavior follower = createBehavior(context);
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
- follower.handleMessage(getTestActor(), new RequestVote(1000, "candidate", 10000, 999));
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
- final Boolean out = new ExpectMsg<Boolean>(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();
+ assertFalse(syncStatus.isInitialSyncDone());
+ assertFalse("append entries reply should be false", reply.isSuccess());
+ }
- assertEquals(false, out);
- }
- };
- }};
+ @Test
+ public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInLog()
+ throws Exception {
+ logStart("testHandleFirstAppendEntries");
+
+ MockRaftActorContext context = createActorContext();
+ context.getReplicatedLog().clear(0,2);
+ context.getReplicatedLog().append(newReplicatedLogEntry(1, 100, "bar"));
+ context.getReplicatedLog().setSnapshotIndex(99);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
+
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+ assertTrue("append entries reply should be true", reply.isSuccess());
+ }
+
+ @Test
+ public void testHandleFirstAppendEntriesWithPrevIndexMinusOneAndReplicatedToAllIndexPresentInSnapshot()
+ throws Exception {
+ logStart("testHandleFirstAppendEntries");
+
+ MockRaftActorContext context = createActorContext();
+ context.getReplicatedLog().clear(0,2);
+ context.getReplicatedLog().setSnapshotIndex(100);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 101, 100, (short) 0);
+
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+ assertTrue("append entries reply should be true", reply.isSuccess());
+ }
+
+ @Test
+ public void testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing()
+ throws Exception {
+ logStart(
+ "testFirstAppendEntriesWithNoPrevIndexAndReplicatedToAllPresentInSnapshotButCalculatedPrevEntryMissing");
+
+ MockRaftActorContext context = createActorContext();
+ context.getReplicatedLog().clear(0,2);
+ context.getReplicatedLog().setSnapshotIndex(100);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 105, "foo"));
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", -1, -1, entries, 105, 100, (short) 0);
+
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+ assertFalse("append entries reply should be false", reply.isSuccess());
}
+ @Test
+ public void testHandleSyncUpAppendEntries() throws Exception {
+ logStart("testHandleSyncUpAppendEntries");
+
+ MockRaftActorContext context = createActorContext();
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
+
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+
+ // Clear all the messages
+ followerActor.underlyingActor().clear();
+
+ context.setLastApplied(101);
+ context.setCommitIndex(101);
+ setLastLogEntry(context, 1, 101,
+ new MockRaftActorContext.MockPayload(""));
+
+ entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+
+ assertTrue(syncStatus.isInitialSyncDone());
+
+ followerActor.underlyingActor().clear();
+
+ // Sending the same message again should not generate another message
+
+ follower.handleMessage(leaderActor, appendEntries);
+
+ syncStatus = MessageCollectorActor.getFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+
+ assertNull(syncStatus);
+
+ }
+
+ @Test
+ public void testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete() throws Exception {
+ logStart("testHandleAppendEntriesLeaderChangedBeforeSyncUpComplete");
+
+ MockRaftActorContext context = createActorContext();
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
+
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+
+ // Clear all the messages
+ followerActor.underlyingActor().clear();
+
+ context.setLastApplied(100);
+ setLastLogEntry(context, 1, 100,
+ new MockRaftActorContext.MockPayload(""));
+
+ entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // leader-2 is becoming the leader now and it says the commitIndex is 45
+ appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+
+ // We get a new message saying initial status is not done
+ assertFalse(syncStatus.isInitialSyncDone());
+
+ }
+
+
+ @Test
+ public void testHandleAppendEntriesLeaderChangedAfterSyncUpComplete() throws Exception {
+ logStart("testHandleAppendEntriesLeaderChangedAfterSyncUpComplete");
+
+ MockRaftActorContext context = createActorContext();
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
+
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ FollowerInitialSyncUpStatus syncStatus = MessageCollectorActor.expectFirstMatching(followerActor,
+ FollowerInitialSyncUpStatus.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+
+ // Clear all the messages
+ followerActor.underlyingActor().clear();
+
+ context.setLastApplied(101);
+ context.setCommitIndex(101);
+ setLastLogEntry(context, 1, 101,
+ new MockRaftActorContext.MockPayload(""));
+
+ entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ appendEntries = new AppendEntries(2, "leader-1", 101, 1, entries, 102, 101, (short)0);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+
+ assertTrue(syncStatus.isInitialSyncDone());
+
+ // Clear all the messages
+ followerActor.underlyingActor().clear();
+
+ context.setLastApplied(100);
+ setLastLogEntry(context, 1, 100,
+ new MockRaftActorContext.MockPayload(""));
+
+ entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // leader-2 is becoming the leader now and it says the commitIndex is 45
+ appendEntries = new AppendEntries(2, "leader-2", 45, 1, entries, 46, 100, (short)0);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+
+ // We get a new message saying initial status is not done
+ assertFalse(syncStatus.isInitialSyncDone());
+
+ }
+
+
/**
* 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()) {{
+ logStart("testHandleAppendEntriesWithNewerCommitIndex");
- RaftActorContext context =
- createActorContext();
+ MockRaftActorContext context = createActorContext();
- context.setLastApplied(100);
- setLastLogEntry((MockRaftActorContext) context, 1, 100,
+ context.setLastApplied(100);
+ setLastLogEntry(context, 1, 100,
new MockRaftActorContext.MockPayload(""));
- ((MockRaftActorContext) context).getReplicatedLog().setSnapshotIndex(99);
-
- List<ReplicatedLogEntry> entries =
- Arrays.asList(
- (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(2, 101,
- new MockRaftActorContext.MockPayload("foo"))
- );
+ context.getReplicatedLog().setSnapshotIndex(99);
- // The new commitIndex is 101
- AppendEntries appendEntries =
- new AppendEntries(2, "leader-1", 100, 1, entries, 101);
+ List<ReplicatedLogEntry> entries = Arrays.<ReplicatedLogEntry>asList(
+ newReplicatedLogEntry(2, 101, "foo"));
- RaftActorBehavior raftBehavior =
- createBehavior(context).handleMessage(getRef(), appendEntries);
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader-1", 100, 1, entries, 101, 100, (short)0);
- assertEquals(101L, context.getLastApplied());
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, appendEntries);
- }};
+ assertEquals("getLastApplied", 101L, context.getLastApplied());
}
/**
* 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 testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm() {
+ logStart("testHandleAppendEntriesSenderPrevLogTermNotSameAsReceiverPrevLogTerm");
- MockRaftActorContext context = (MockRaftActorContext)
- createActorContext();
+ MockRaftActorContext context = createActorContext();
- // First set the receivers term to lower number
- context.getTermInformation().update(95, "test");
+ // First set the receivers term to lower number
+ context.getTermInformation().update(95, "test");
- // Set the last log entry term for the receiver to be greater than
- // what we will be sending as the prevLogTerm in AppendEntries
- MockRaftActorContext.SimpleReplicatedLog mockReplicatedLog =
- setLastLogEntry(context, 20, 0, new MockRaftActorContext.MockPayload(""));
+ // 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", 0, 0, null, 101, -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);
+ follower = createBehavior(context);
- RaftActorBehavior behavior = createBehavior(context);
+ RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
- // Send an unknown message so that the state of the RaftActor remains unchanged
- RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
+ Assert.assertSame(follower, newBehavior);
- RaftActorBehavior raftBehavior =
- behavior.handleMessage(getRef(), appendEntries);
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
+ AppendEntriesReply.class);
- assertEquals(expected, raftBehavior);
-
- // Also expect an AppendEntriesReply to be sent where success is false
- final Boolean out = new ExpectMsg<Boolean>(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(false, out);
-
-
- }};
+ assertEquals("isSuccess", false, reply.isSuccess());
}
-
-
/**
* 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
+ * entries received in appendEntries.
*/
@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, new MockRaftActorContext.MockPayload("zero")));
- log.append(
- new MockRaftActorContext.MockReplicatedLogEntry(1, 1, new MockRaftActorContext.MockPayload("one")));
- log.append(
- new MockRaftActorContext.MockReplicatedLogEntry(1, 2, new MockRaftActorContext.MockPayload("two")));
-
- context.setReplicatedLog(log);
-
- // Prepare the entries to be sent with AppendEntries
- List<ReplicatedLogEntry> entries = new ArrayList<>();
- entries.add(
- new MockRaftActorContext.MockReplicatedLogEntry(1, 3, new MockRaftActorContext.MockPayload("three")));
- entries.add(
- new MockRaftActorContext.MockReplicatedLogEntry(1, 4, new MockRaftActorContext.MockPayload("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, 4);
-
- RaftActorBehavior behavior = createBehavior(context);
-
- // Send an unknown message so that the state of the RaftActor remains unchanged
- RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
-
- RaftActorBehavior raftBehavior =
- behavior.handleMessage(getRef(), appendEntries);
-
- assertEquals(expected, raftBehavior);
- assertEquals(5, log.last().getIndex() + 1);
- assertNotNull(log.get(3));
- assertNotNull(log.get(4));
-
- // Also expect an AppendEntriesReply to be sent where success is false
- final Boolean out = new ExpectMsg<Boolean>(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();
+ public void testHandleAppendEntriesAddNewEntries() {
+ logStart("testHandleAppendEntriesAddNewEntries");
- assertEquals(true, out);
+ MockRaftActorContext context = 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(newReplicatedLogEntry(1, 0, "zero"));
+ log.append(newReplicatedLogEntry(1, 1, "one"));
+ log.append(newReplicatedLogEntry(1, 2, "two"));
+
+ context.setReplicatedLog(log);
+
+ // Prepare the entries to be sent with AppendEntries
+ List<ReplicatedLogEntry> entries = new ArrayList<>();
+ entries.add(newReplicatedLogEntry(1, 3, "three"));
+ entries.add(newReplicatedLogEntry(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
+ short leaderPayloadVersion = 10;
+ String leaderId = "leader-1";
+ AppendEntries appendEntries = new AppendEntries(1, leaderId, 2, 1, entries, 4, -1, leaderPayloadVersion);
+ follower = createBehavior(context);
+ RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
+
+ Assert.assertSame(follower, newBehavior);
+
+ assertEquals("Next index", 5, log.last().getIndex() + 1);
+ assertEquals("Entry 3", entries.get(0), log.get(3));
+ assertEquals("Entry 4", entries.get(1), log.get(4));
+
+ assertEquals("getLeaderPayloadVersion", leaderPayloadVersion, newBehavior.getLeaderPayloadVersion());
+ assertEquals("getLeaderId", leaderId, newBehavior.getLeaderId());
+
+ expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
+ }
/**
* 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
+ * log and then adding in the new entries sent with the AppendEntries message.
*/
@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, new MockRaftActorContext.MockPayload("zero")));
- log.append(
- new MockRaftActorContext.MockReplicatedLogEntry(1, 1, new MockRaftActorContext.MockPayload("one")));
- log.append(
- new MockRaftActorContext.MockReplicatedLogEntry(1, 2, new MockRaftActorContext.MockPayload("two")));
-
- context.setReplicatedLog(log);
-
- // Prepare the entries to be sent with AppendEntries
- List<ReplicatedLogEntry> entries = new ArrayList<>();
- entries.add(
- new MockRaftActorContext.MockReplicatedLogEntry(2, 2, new MockRaftActorContext.MockPayload("two-1")));
- entries.add(
- new MockRaftActorContext.MockReplicatedLogEntry(2, 3, new MockRaftActorContext.MockPayload("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, 3);
-
- RaftActorBehavior behavior = createBehavior(context);
-
- // Send an unknown message so that the state of the RaftActor remains unchanged
- RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
-
- RaftActorBehavior raftBehavior =
- behavior.handleMessage(getRef(), appendEntries);
-
- assertEquals(expected, raftBehavior);
-
- // 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));
-
- assertEquals("one", log.get(1).getData().toString());
-
- // Check that the entry at index 2 has the new data
- assertEquals("two-1", log.get(2).getData().toString());
-
- assertEquals("three", log.get(3).getData().toString());
-
- assertNotNull(log.get(3));
-
- // Also expect an AppendEntriesReply to be sent where success is false
- final Boolean out = new ExpectMsg<Boolean>(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();
+ public void testHandleAppendEntriesCorrectReceiverLogEntries() {
+ logStart("testHandleAppendEntriesCorrectReceiverLogEntries");
+
+ MockRaftActorContext context = 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(newReplicatedLogEntry(1, 0, "zero"));
+ log.append(newReplicatedLogEntry(1, 1, "one"));
+ log.append(newReplicatedLogEntry(1, 2, "two"));
+
+ context.setReplicatedLog(log);
+
+ // Prepare the entries to be sent with AppendEntries
+ List<ReplicatedLogEntry> entries = new ArrayList<>();
+ entries.add(newReplicatedLogEntry(2, 2, "two-1"));
+ entries.add(newReplicatedLogEntry(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, entries, 3, -1, (short)0);
+
+ follower = createBehavior(context);
+
+ RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
+
+ Assert.assertSame(follower, newBehavior);
+
+ // 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("Next index", 4, log.last().getIndex() + 1);
+ //assertEquals("Entry 2", entries.get(0), log.get(2));
+
+ assertEquals("Entry 1 data", "one", log.get(1).getData().toString());
+
+ // Check that the entry at index 2 has the new data
+ assertEquals("Entry 2", entries.get(0), log.get(2));
+
+ assertEquals("Entry 3", entries.get(1), log.get(3));
+
+ expectAndVerifyAppendEntriesReply(2, true, context.getId(), 2, 3);
+ }
+
+ @Test
+ public void testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot() {
+ logStart("testHandleAppendEntriesWhenOutOfSyncLogDetectedRequestForceInstallSnapshot");
+
+ MockRaftActorContext context = 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(newReplicatedLogEntry(1, 0, "zero"));
+ log.append(newReplicatedLogEntry(1, 1, "one"));
+ log.append(newReplicatedLogEntry(1, 2, "two"));
+
+ context.setReplicatedLog(log);
+
+ // Prepare the entries to be sent with AppendEntries
+ List<ReplicatedLogEntry> entries = new ArrayList<>();
+ entries.add(newReplicatedLogEntry(2, 2, "two-1"));
+ entries.add(newReplicatedLogEntry(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, entries, 3, -1, (short)0);
+
+ context.setRaftPolicy(createRaftPolicy(false, true));
+ follower = createBehavior(context);
+
+ RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
+
+ Assert.assertSame(follower, newBehavior);
+
+ expectAndVerifyAppendEntriesReply(2, false, context.getId(), 1, 2, true);
+ }
+
+ @Test
+ public void testHandleAppendEntriesPreviousLogEntryMissing() {
+ logStart("testHandleAppendEntriesPreviousLogEntryMissing");
+
+ final MockRaftActorContext context = createActorContext();
+
+ // Prepare the receivers log
+ MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
+ log.append(newReplicatedLogEntry(1, 0, "zero"));
+ log.append(newReplicatedLogEntry(1, 1, "one"));
+ log.append(newReplicatedLogEntry(1, 2, "two"));
+
+ context.setReplicatedLog(log);
+
+ // Prepare the entries to be sent with AppendEntries
+ List<ReplicatedLogEntry> entries = new ArrayList<>();
+ entries.add(newReplicatedLogEntry(1, 4, "four"));
+
+ AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, -1, (short)0);
+
+ follower = createBehavior(context);
+
+ RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
+
+ Assert.assertSame(follower, newBehavior);
+
+ expectAndVerifyAppendEntriesReply(1, false, context.getId(), 1, 2);
+ }
+
+ @Test
+ public void testHandleAppendEntriesWithExistingLogEntry() {
+ logStart("testHandleAppendEntriesWithExistingLogEntry");
+
+ MockRaftActorContext context = createActorContext();
+
+ context.getTermInformation().update(1, "test");
+
+ // Prepare the receivers log
+ MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
+ log.append(newReplicatedLogEntry(1, 0, "zero"));
+ log.append(newReplicatedLogEntry(1, 1, "one"));
+
+ context.setReplicatedLog(log);
+
+ // Send the last entry again.
+ List<ReplicatedLogEntry> entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"));
+
+ follower = createBehavior(context);
+
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 1, -1, (short)0));
+
+ assertEquals("Next index", 2, log.last().getIndex() + 1);
+ assertEquals("Entry 1", entries.get(0), log.get(1));
+
+ expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 1);
+
+ // Send the last entry again and also a new one.
+
+ entries = Arrays.asList(newReplicatedLogEntry(1, 1, "one"), newReplicatedLogEntry(1, 2, "two"));
+
+ leaderActor.underlyingActor().clear();
+ follower.handleMessage(leaderActor, new AppendEntries(1, "leader", 0, 1, entries, 2, -1, (short)0));
+
+ assertEquals("Next index", 3, log.last().getIndex() + 1);
+ assertEquals("Entry 1", entries.get(0), log.get(1));
+ assertEquals("Entry 2", entries.get(1), log.get(2));
+
+ expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 2);
+ }
+
+ @Test
+ public void testHandleAppendEntriesAfterInstallingSnapshot() {
+ logStart("testHandleAppendAfterInstallingSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+
+ // Prepare the receivers log
+ MockRaftActorContext.SimpleReplicatedLog log = new MockRaftActorContext.SimpleReplicatedLog();
+
+ // Set up a log as if it has been snapshotted
+ log.setSnapshotIndex(3);
+ log.setSnapshotTerm(1);
- assertEquals(true, out);
+ context.setReplicatedLog(log);
+ // Prepare the entries to be sent with AppendEntries
+ List<ReplicatedLogEntry> entries = new ArrayList<>();
+ entries.add(newReplicatedLogEntry(1, 4, "four"));
- }};
+ AppendEntries appendEntries = new AppendEntries(1, "leader", 3, 1, entries, 4, 3, (short)0);
+
+ follower = createBehavior(context);
+
+ RaftActorBehavior newBehavior = follower.handleMessage(leaderActor, appendEntries);
+
+ Assert.assertSame(follower, newBehavior);
+
+ expectAndVerifyAppendEntriesReply(1, true, context.getId(), 1, 4);
}
/**
* This test verifies that when InstallSnapshot is received by
* the follower its applied correctly.
- *
- * @throws Exception
*/
@Test
public void testHandleInstallSnapshot() throws Exception {
- JavaTestKit javaTestKit = new JavaTestKit(getSystem()) {{
-
- ActorRef leaderActor = getSystem().actorOf(Props.create(
- MessageCollectorActor.class));
-
- MockRaftActorContext context = (MockRaftActorContext)
- createActorContext(getRef());
-
- Follower follower = (Follower)createBehavior(context);
-
- HashMap<String, String> followerSnapshot = new HashMap<>();
- followerSnapshot.put("1", "A");
- followerSnapshot.put("2", "B");
- followerSnapshot.put("3", "C");
-
- ByteString bsSnapshot = toByteString(followerSnapshot);
- ByteString chunkData = ByteString.EMPTY;
- int offset = 0;
- int snapshotLength = bsSnapshot.size();
- int i = 1;
-
- do {
- chunkData = getNextChunk(bsSnapshot, offset);
- final InstallSnapshot installSnapshot =
- new InstallSnapshot(1, "leader-1", i, 1,
- chunkData, i, 3);
- follower.handleMessage(leaderActor, installSnapshot);
- offset = offset + 50;
- i++;
- } while ((offset+50) < snapshotLength);
-
- final InstallSnapshot installSnapshot3 = new InstallSnapshot(1, "leader-1", 3, 1, chunkData, 3, 3);
- follower.handleMessage(leaderActor, installSnapshot3);
-
- String[] matches = new ReceiveWhile<String>(String.class, duration("2 seconds")) {
- @Override
- protected String match(Object o) throws Exception {
- if (o instanceof ApplySnapshot) {
- ApplySnapshot as = (ApplySnapshot)o;
- if (as.getSnapshot().getLastIndex() != installSnapshot3.getLastIncludedIndex()) {
- return "applySnapshot-lastIndex-mismatch";
- }
- if (as.getSnapshot().getLastAppliedTerm() != installSnapshot3.getLastIncludedTerm()) {
- return "applySnapshot-lastAppliedTerm-mismatch";
- }
- if (as.getSnapshot().getLastAppliedIndex() != installSnapshot3.getLastIncludedIndex()) {
- return "applySnapshot-lastAppliedIndex-mismatch";
- }
- if (as.getSnapshot().getLastTerm() != installSnapshot3.getLastIncludedTerm()) {
- return "applySnapshot-lastTerm-mismatch";
- }
- return "applySnapshot";
- }
-
- return "ignoreCase";
- }
- }.get();
+ logStart("testHandleInstallSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+ context.getTermInformation().update(1, "leader");
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+ int offset = 0;
+ int snapshotLength = bsSnapshot.size();
+ int chunkSize = 50;
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
+ int lastIncludedIndex = 1;
+ int chunkIndex = 1;
+ InstallSnapshot lastInstallSnapshot = null;
+
+ for (int i = 0; i < totalChunks; i++) {
+ byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
+ lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
+ chunkData, chunkIndex, totalChunks);
+ follower.handleMessage(leaderActor, lastInstallSnapshot);
+ offset = offset + 50;
+ lastIncludedIndex++;
+ chunkIndex++;
+ }
- String applySnapshotMatch = "";
- for (String reply: matches) {
- if (reply.startsWith("applySnapshot")) {
- applySnapshotMatch = reply;
- }
+ ApplySnapshot applySnapshot = MessageCollectorActor.expectFirstMatching(followerActor,
+ ApplySnapshot.class);
+ Snapshot snapshot = applySnapshot.getSnapshot();
+ assertNotNull(lastInstallSnapshot);
+ assertEquals("getLastIndex", lastInstallSnapshot.getLastIncludedIndex(), snapshot.getLastIndex());
+ assertEquals("getLastIncludedTerm", lastInstallSnapshot.getLastIncludedTerm(),
+ snapshot.getLastAppliedTerm());
+ assertEquals("getLastAppliedIndex", lastInstallSnapshot.getLastIncludedIndex(),
+ snapshot.getLastAppliedIndex());
+ assertEquals("getLastTerm", lastInstallSnapshot.getLastIncludedTerm(), snapshot.getLastTerm());
+ assertEquals("getState type", ByteState.class, snapshot.getState().getClass());
+ Assert.assertArrayEquals("getState", bsSnapshot.toByteArray(), ((ByteState)snapshot.getState()).getBytes());
+ assertEquals("getElectionTerm", 1, snapshot.getElectionTerm());
+ assertEquals("getElectionVotedFor", "leader", snapshot.getElectionVotedFor());
+ applySnapshot.getCallback().onSuccess();
+
+ List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
+ leaderActor, InstallSnapshotReply.class);
+ assertEquals("InstallSnapshotReply count", totalChunks, replies.size());
+
+ chunkIndex = 1;
+ for (InstallSnapshotReply reply: replies) {
+ assertEquals("getChunkIndex", chunkIndex++, reply.getChunkIndex());
+ assertEquals("getTerm", 1, reply.getTerm());
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
+ }
+
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
+ }
+
+
+ /**
+ * Verify that when an AppendEntries is sent to a follower during a snapshot install
+ * the Follower short-circuits the processing of the AppendEntries message.
+ */
+ @Test
+ public void testReceivingAppendEntriesDuringInstallSnapshot() throws Exception {
+ logStart("testReceivingAppendEntriesDuringInstallSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+ int snapshotLength = bsSnapshot.size();
+ int chunkSize = 50;
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
+ int lastIncludedIndex = 1;
+
+ // Check that snapshot installation is not in progress
+ assertNull(follower.getSnapshotTracker());
+
+ // Make sure that we have more than 1 chunk to send
+ assertTrue(totalChunks > 1);
+
+ // Send an install snapshot with the first chunk to start the process of installing a snapshot
+ byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
+ follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
+ chunkData, 1, totalChunks));
+
+ // Check if snapshot installation is in progress now
+ assertNotNull(follower.getSnapshotTracker());
+
+ // Send an append entry
+ AppendEntries appendEntries = new AppendEntries(1, "leader", 1, 1,
+ Arrays.asList(newReplicatedLogEntry(2, 1, "3")), 2, -1, (short)1);
+
+ follower.handleMessage(leaderActor, appendEntries);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getLogLastIndex", context.getReplicatedLog().lastIndex(), reply.getLogLastIndex());
+ assertEquals("getLogLastTerm", context.getReplicatedLog().lastTerm(), reply.getLogLastTerm());
+ assertEquals("getTerm", context.getTermInformation().getCurrentTerm(), reply.getTerm());
+
+ assertNotNull(follower.getSnapshotTracker());
+ }
+
+ @Test
+ public void testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader() throws Exception {
+ logStart("testReceivingAppendEntriesDuringInstallSnapshotFromDifferentLeader");
+
+ MockRaftActorContext context = createActorContext();
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+ int snapshotLength = bsSnapshot.size();
+ int chunkSize = 50;
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
+ int lastIncludedIndex = 1;
+
+ // Check that snapshot installation is not in progress
+ assertNull(follower.getSnapshotTracker());
+
+ // Make sure that we have more than 1 chunk to send
+ assertTrue(totalChunks > 1);
+
+ // Send an install snapshot with the first chunk to start the process of installing a snapshot
+ byte[] chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
+ follower.handleMessage(leaderActor, new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
+ chunkData, 1, totalChunks));
+
+ // Check if snapshot installation is in progress now
+ assertNotNull(follower.getSnapshotTracker());
+
+ // Send appendEntries with a new term and leader.
+ AppendEntries appendEntries = new AppendEntries(2, "new-leader", 1, 1,
+ Arrays.asList(newReplicatedLogEntry(2, 2, "3")), 2, -1, (short)1);
+
+ follower.handleMessage(leaderActor, appendEntries);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+ assertEquals("getLogLastIndex", 2, reply.getLogLastIndex());
+ assertEquals("getLogLastTerm", 2, reply.getLogLastTerm());
+ assertEquals("getTerm", 2, reply.getTerm());
+
+ assertNull(follower.getSnapshotTracker());
+ }
+
+ @Test
+ public void testInitialSyncUpWithHandleInstallSnapshotFollowedByAppendEntries() throws Exception {
+ logStart("testInitialSyncUpWithHandleInstallSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+ context.setCommitIndex(-1);
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+ int offset = 0;
+ int snapshotLength = bsSnapshot.size();
+ int chunkSize = 50;
+ int totalChunks = snapshotLength / chunkSize + (snapshotLength % chunkSize > 0 ? 1 : 0);
+ int lastIncludedIndex = 1;
+ int chunkIndex = 1;
+ InstallSnapshot lastInstallSnapshot = null;
+
+ for (int i = 0; i < totalChunks; i++) {
+ byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
+ lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
+ chunkData, chunkIndex, totalChunks);
+ follower.handleMessage(leaderActor, lastInstallSnapshot);
+ offset = offset + 50;
+ lastIncludedIndex++;
+ chunkIndex++;
+ }
+
+ FollowerInitialSyncUpStatus syncStatus =
+ MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+
+ assertFalse(syncStatus.isInitialSyncDone());
+
+ // Clear all the messages
+ followerActor.underlyingActor().clear();
+
+ context.setLastApplied(101);
+ context.setCommitIndex(101);
+ setLastLogEntry(context, 1, 101,
+ new MockRaftActorContext.MockPayload(""));
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(2, 101, "foo"));
+
+ // The new commitIndex is 101
+ AppendEntries appendEntries = new AppendEntries(2, "leader", 101, 1, entries, 102, 101, (short)0);
+ follower.handleMessage(leaderActor, appendEntries);
+
+ syncStatus = MessageCollectorActor.expectFirstMatching(followerActor, FollowerInitialSyncUpStatus.class);
+
+ assertTrue(syncStatus.isInitialSyncDone());
+ }
+
+ @Test
+ public void testHandleOutOfSequenceInstallSnapshot() throws Exception {
+ logStart("testHandleOutOfSequenceInstallSnapshot");
+
+ MockRaftActorContext context = createActorContext();
+
+ follower = createBehavior(context);
+
+ ByteString bsSnapshot = createSnapshot();
+
+ InstallSnapshot installSnapshot = new InstallSnapshot(1, "leader", 3, 1,
+ getNextChunk(bsSnapshot, 10, 50), 3, 3);
+ follower.handleMessage(leaderActor, installSnapshot);
+
+ InstallSnapshotReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
+ InstallSnapshotReply.class);
+
+ assertEquals("isSuccess", false, reply.isSuccess());
+ assertEquals("getChunkIndex", -1, reply.getChunkIndex());
+ assertEquals("getTerm", 1, reply.getTerm());
+ assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
+
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
+ }
+
+ @Test
+ public void testFollowerSchedulesElectionTimeoutImmediatelyWhenItHasNoPeers() {
+ MockRaftActorContext context = createActorContext();
+
+ Stopwatch stopwatch = Stopwatch.createStarted();
+
+ follower = createBehavior(context);
+
+ TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
+
+ long elapsed = stopwatch.elapsed(TimeUnit.MILLISECONDS);
+
+ assertTrue(elapsed < context.getConfigParams().getElectionTimeOutInterval().toMillis());
+
+ RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
+ assertTrue("Expected Candidate", newBehavior instanceof Candidate);
+ }
+
+ @Test
+ public void testFollowerSchedulesElectionIfAutomaticElectionsAreDisabled() {
+ MockRaftActorContext context = createActorContext();
+ context.setConfigParams(new DefaultConfigParamsImpl() {
+ @Override
+ public FiniteDuration getElectionTimeOutInterval() {
+ return FiniteDuration.apply(100, TimeUnit.MILLISECONDS);
}
+ });
- assertEquals("applySnapshot", applySnapshotMatch);
+ context.setRaftPolicy(createRaftPolicy(false, false));
- Object messages = executeLocalOperation(leaderActor, "get-all-messages");
+ follower = createBehavior(context);
- assertNotNull(messages);
- assertTrue(messages instanceof List);
- List<Object> listMessages = (List<Object>) messages;
+ TimeoutNow timeoutNow = MessageCollectorActor.expectFirstMatching(followerActor, TimeoutNow.class);
+ RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), timeoutNow);
+ assertSame("handleMessage result", follower, newBehavior);
+ }
- int installSnapshotReplyReceivedCount = 0;
- for (Object message: listMessages) {
- if (message instanceof InstallSnapshotReply) {
- ++installSnapshotReplyReceivedCount;
- }
+ @Test
+ public void testFollowerSchedulesElectionIfNonVoting() {
+ MockRaftActorContext context = createActorContext();
+ context.updatePeerIds(new ServerConfigurationPayload(Arrays.asList(new ServerInfo(context.getId(), false))));
+ ((DefaultConfigParamsImpl)context.getConfigParams()).setHeartBeatInterval(
+ FiniteDuration.apply(100, TimeUnit.MILLISECONDS));
+ ((DefaultConfigParamsImpl)context.getConfigParams()).setElectionTimeoutFactor(1);
+
+ follower = new Follower(context, "leader", (short)1);
+
+ ElectionTimeout electionTimeout = MessageCollectorActor.expectFirstMatching(followerActor,
+ ElectionTimeout.class);
+ RaftActorBehavior newBehavior = follower.handleMessage(ActorRef.noSender(), electionTimeout);
+ assertSame("handleMessage result", follower, newBehavior);
+ assertNull("Expected null leaderId", follower.getLeaderId());
+ }
+
+ @Test
+ public void testElectionScheduledWhenAnyRaftRPCReceived() {
+ MockRaftActorContext context = createActorContext();
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, new RaftRPC() {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public long getTerm() {
+ return 100;
}
+ });
+ verify(follower).scheduleElection(any(FiniteDuration.class));
+ }
- assertEquals(3, installSnapshotReplyReceivedCount);
+ @Test
+ public void testElectionNotScheduledWhenNonRaftRPCMessageReceived() {
+ MockRaftActorContext context = createActorContext();
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, "non-raft-rpc");
+ verify(follower, never()).scheduleElection(any(FiniteDuration.class));
+ }
- }};
+ @Test
+ public void testCaptureSnapshotOnLastEntryInAppendEntries() throws Exception {
+ String id = "testCaptureSnapshotOnLastEntryInAppendEntries";
+ logStart(id);
+
+ InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setSnapshotBatchCount(2);
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
+ RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
+ Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
+ .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
+ TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(id);
+ InMemoryJournal.addDeleteMessagesCompleteLatch(id);
+ InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"));
+
+ AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 1, -1, (short)0);
+
+ followerActorRef.tell(appendEntries, leaderActor);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+
+ final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
+
+ InMemoryJournal.waitForDeleteMessagesComplete(id);
+ InMemoryJournal.waitForWriteMessagesComplete(id);
+ // We expect the ApplyJournalEntries for index 1 to remain in the persisted log b/c it's still queued for
+ // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
+ // This is OK - on recovery it will be a no-op since index 1 has already been applied.
+ List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
+ assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
+ assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
+ assertEquals("ApplyJournalEntries index", 1, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
+
+ assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
+ assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
+ assertEquals("Snapshot getLastAppliedIndex", 1, snapshot.getLastAppliedIndex());
+ assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
+ assertEquals("Snapshot getLastIndex", 1, snapshot.getLastIndex());
+ assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData()),
+ MockRaftActor.fromState(snapshot.getState()));
}
- public Object executeLocalOperation(ActorRef actor, Object message) throws Exception {
- return MessageCollectorActor.getAllMessages(actor);
+ @Test
+ public void testCaptureSnapshotOnMiddleEntryInAppendEntries() throws Exception {
+ String id = "testCaptureSnapshotOnMiddleEntryInAppendEntries";
+ logStart(id);
+
+ InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setSnapshotBatchCount(2);
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
+ RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
+ Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
+ .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
+ TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(id);
+ InMemoryJournal.addDeleteMessagesCompleteLatch(id);
+ InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
+ newReplicatedLogEntry(1, 2, "three"));
+
+ AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 2, -1, (short)0);
+
+ followerActorRef.tell(appendEntries, leaderActor);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+
+ final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
+
+ InMemoryJournal.waitForDeleteMessagesComplete(id);
+ InMemoryJournal.waitForWriteMessagesComplete(id);
+ // We expect the ApplyJournalEntries for index 2 to remain in the persisted log b/c it's still queued for
+ // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
+ // This is OK - on recovery it will be a no-op since index 2 has already been applied.
+ List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
+ assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
+ assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
+ assertEquals("ApplyJournalEntries index", 2, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
+
+ assertEquals("Snapshot unapplied size", 0, snapshot.getUnAppliedEntries().size());
+ assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
+ assertEquals("Snapshot getLastAppliedIndex", 2, snapshot.getLastAppliedIndex());
+ assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
+ assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
+ assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
+ entries.get(2).getData()), MockRaftActor.fromState(snapshot.getState()));
+
+ assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
+ assertEquals("Snapshot index", 2, followerRaftActor.get().getReplicatedLog().getSnapshotIndex());
+
+ // Reinstate the actor from persistence
+
+ actorFactory.killActor(followerActorRef, new JavaTestKit(getSystem()));
+
+ followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ assertEquals("Journal size", 0, followerRaftActor.get().getReplicatedLog().size());
+ assertEquals("Last index", 2, followerRaftActor.get().getReplicatedLog().lastIndex());
+ assertEquals("Last applied index", 2, followerRaftActor.get().getRaftActorContext().getLastApplied());
+ assertEquals("Commit index", 2, followerRaftActor.get().getRaftActorContext().getCommitIndex());
+ assertEquals("State", ImmutableList.of(entries.get(0).getData(), entries.get(1).getData(),
+ entries.get(2).getData()), followerRaftActor.get().getState());
+ }
+
+ @Test
+ public void testCaptureSnapshotOnAppendEntriesWithUnapplied() throws Exception {
+ String id = "testCaptureSnapshotOnAppendEntriesWithUnapplied";
+ logStart(id);
+
+ InMemoryJournal.addEntry(id, 1, new UpdateElectionTerm(1, null));
+
+ DefaultConfigParamsImpl config = new DefaultConfigParamsImpl();
+ config.setSnapshotBatchCount(1);
+ config.setCustomRaftPolicyImplementationClass(DisableElectionsRaftPolicy.class.getName());
+
+ final AtomicReference<MockRaftActor> followerRaftActor = new AtomicReference<>();
+ RaftActorSnapshotCohort snapshotCohort = newRaftActorSnapshotCohort(followerRaftActor);
+ Builder builder = MockRaftActor.builder().persistent(Optional.of(true)).id(id)
+ .peerAddresses(ImmutableMap.of("leader", "")).config(config).snapshotCohort(snapshotCohort);
+ TestActorRef<MockRaftActor> followerActorRef = actorFactory.createTestActor(builder.props()
+ .withDispatcher(Dispatchers.DefaultDispatcherId()), id);
+ followerRaftActor.set(followerActorRef.underlyingActor());
+ followerRaftActor.get().waitForInitializeBehaviorComplete();
+
+ InMemorySnapshotStore.addSnapshotSavedLatch(id);
+ InMemoryJournal.addDeleteMessagesCompleteLatch(id);
+ InMemoryJournal.addWriteMessagesCompleteLatch(id, 1, ApplyJournalEntries.class);
+
+ List<ReplicatedLogEntry> entries = Arrays.asList(
+ newReplicatedLogEntry(1, 0, "one"), newReplicatedLogEntry(1, 1, "two"),
+ newReplicatedLogEntry(1, 2, "three"));
+
+ AppendEntries appendEntries = new AppendEntries(1, "leader", -1, -1, entries, 0, -1, (short)0);
+
+ followerActorRef.tell(appendEntries, leaderActor);
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
+
+ final Snapshot snapshot = InMemorySnapshotStore.waitForSavedSnapshot(id, Snapshot.class);
+
+ InMemoryJournal.waitForDeleteMessagesComplete(id);
+ InMemoryJournal.waitForWriteMessagesComplete(id);
+ // We expect the ApplyJournalEntries for index 0 to remain in the persisted log b/c it's still queued for
+ // persistence by the time we initiate capture so the last persisted journal sequence number doesn't include it.
+ // This is OK - on recovery it will be a no-op since index 0 has already been applied.
+ List<Object> journalEntries = InMemoryJournal.get(id, Object.class);
+ assertEquals("Persisted journal entries size: " + journalEntries, 1, journalEntries.size());
+ assertEquals("Persisted journal entry type", ApplyJournalEntries.class, journalEntries.get(0).getClass());
+ assertEquals("ApplyJournalEntries index", 0, ((ApplyJournalEntries)journalEntries.get(0)).getToIndex());
+
+ assertEquals("Snapshot unapplied size", 2, snapshot.getUnAppliedEntries().size());
+ assertEquals("Snapshot unapplied entry index", 1, snapshot.getUnAppliedEntries().get(0).getIndex());
+ assertEquals("Snapshot unapplied entry index", 2, snapshot.getUnAppliedEntries().get(1).getIndex());
+ assertEquals("Snapshot getLastAppliedTerm", 1, snapshot.getLastAppliedTerm());
+ assertEquals("Snapshot getLastAppliedIndex", 0, snapshot.getLastAppliedIndex());
+ assertEquals("Snapshot getLastTerm", 1, snapshot.getLastTerm());
+ assertEquals("Snapshot getLastIndex", 2, snapshot.getLastIndex());
+ assertEquals("Snapshot state", ImmutableList.of(entries.get(0).getData()),
+ MockRaftActor.fromState(snapshot.getState()));
}
- public ByteString getNextChunk (ByteString bs, int offset){
+ @SuppressWarnings("checkstyle:IllegalCatch")
+ private RaftActorSnapshotCohort newRaftActorSnapshotCohort(final AtomicReference<MockRaftActor> followerRaftActor) {
+ RaftActorSnapshotCohort snapshotCohort = new RaftActorSnapshotCohort() {
+ @Override
+ public void createSnapshot(ActorRef actorRef, java.util.Optional<OutputStream> installSnapshotStream) {
+ try {
+ actorRef.tell(new CaptureSnapshotReply(new MockSnapshotState(followerRaftActor.get().getState()),
+ installSnapshotStream), actorRef);
+ } catch (Exception e) {
+ Throwables.propagate(e);
+ }
+ }
+
+ @Override
+ public void applySnapshot(State snapshotState) {
+ }
+
+ @Override
+ public State deserializeSnapshot(ByteSource snapshotBytes) {
+ throw new UnsupportedOperationException();
+ }
+ };
+ return snapshotCohort;
+ }
+
+ public byte[] getNextChunk(ByteString bs, int offset, int chunkSize) {
int snapshotLength = bs.size();
int start = offset;
- int size = 50;
- if (50 > snapshotLength) {
+ int size = chunkSize;
+ if (chunkSize > snapshotLength) {
size = snapshotLength;
} else {
- if ((start + 50) > snapshotLength) {
+ if (start + chunkSize > snapshotLength) {
size = snapshotLength - start;
}
}
- return bs.substring(start, start + size);
- }
-
- private ByteString toByteString(Map<String, String> state) {
- ByteArrayOutputStream b = null;
- ObjectOutputStream o = null;
- try {
- try {
- b = new ByteArrayOutputStream();
- o = new ObjectOutputStream(b);
- o.writeObject(state);
- byte[] snapshotBytes = b.toByteArray();
- return ByteString.copyFrom(snapshotBytes);
- } finally {
- if (o != null) {
- o.flush();
- o.close();
- }
- if (b != null) {
- b.close();
- }
- }
- } catch (IOException e) {
- org.junit.Assert.fail("IOException in converting Hashmap to Bytestring:" + e);
- }
- return null;
+
+ byte[] nextChunk = new byte[size];
+ bs.copyTo(nextChunk, start, 0, size);
+ return nextChunk;
+ }
+
+ private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
+ String expFollowerId, long expLogLastTerm, long expLogLastIndex) {
+ expectAndVerifyAppendEntriesReply(expTerm, expSuccess, expFollowerId, expLogLastTerm, expLogLastIndex, false);
+ }
+
+ private void expectAndVerifyAppendEntriesReply(int expTerm, boolean expSuccess,
+ String expFollowerId, long expLogLastTerm, long expLogLastIndex,
+ boolean expForceInstallSnapshot) {
+
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(leaderActor,
+ AppendEntriesReply.class);
+
+ assertEquals("isSuccess", expSuccess, reply.isSuccess());
+ assertEquals("getTerm", expTerm, reply.getTerm());
+ assertEquals("getFollowerId", expFollowerId, reply.getFollowerId());
+ assertEquals("getLogLastTerm", expLogLastTerm, reply.getLogLastTerm());
+ assertEquals("getLogLastIndex", expLogLastIndex, reply.getLogLastIndex());
+ assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion());
+ assertEquals("isForceInstallSnapshot", expForceInstallSnapshot, reply.isForceInstallSnapshot());
+ }
+
+
+ private static ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
+ return new SimpleReplicatedLogEntry(index, term,
+ new MockRaftActorContext.MockPayload(data));
+ }
+
+ private ByteString createSnapshot() {
+ HashMap<String, String> followerSnapshot = new HashMap<>();
+ followerSnapshot.put("1", "A");
+ followerSnapshot.put("2", "B");
+ followerSnapshot.put("3", "C");
+
+ return toByteString(followerSnapshot);
+ }
+
+ @Override
+ protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(MockRaftActorContext actorContext,
+ ActorRef actorRef, RaftRPC rpc) throws Exception {
+ super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
+
+ String expVotedFor = rpc instanceof RequestVote ? ((RequestVote)rpc).getCandidateId() : null;
+ assertEquals("New votedFor", expVotedFor, actorContext.getTermInformation().getVotedFor());
+ }
+
+ @Override
+ protected void handleAppendEntriesAddSameEntryToLogReply(final TestActorRef<MessageCollectorActor> replyActor)
+ throws Exception {
+ AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(replyActor, AppendEntriesReply.class);
+ assertEquals("isSuccess", true, reply.isSuccess());
}
}