import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Matchers.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
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 org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
import scala.concurrent.duration.FiniteDuration;
-public class FollowerTest extends AbstractRaftActorBehaviorTest {
+public class FollowerTest extends AbstractRaftActorBehaviorTest<Follower> {
private final TestActorRef<MessageCollectorActor> followerActor = actorFactory.createTestActor(
Props.create(MessageCollectorActor.class), actorFactory.generateActorId("follower"));
private final TestActorRef<MessageCollectorActor> leaderActor = actorFactory.createTestActor(
Props.create(MessageCollectorActor.class), actorFactory.generateActorId("leader"));
- private RaftActorBehavior follower;
+ private Follower follower;
private final short payloadVersion = 5;
}
@Override
- protected RaftActorBehavior createBehavior(RaftActorContext actorContext) {
- return new Follower(actorContext);
+ protected Follower createBehavior(RaftActorContext actorContext) {
+ return spy(new Follower(actorContext));
}
@Override
follower = new Follower(createActorContext());
- RaftActorBehavior raftBehavior = follower.handleMessage(followerActor, new ElectionTimeout());
+ RaftActorBehavior raftBehavior = follower.handleMessage(followerActor, ElectionTimeout.INSTANCE);
assertTrue(raftBehavior instanceof Candidate);
}
public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull(){
logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNull");
- RaftActorContext context = createActorContext();
+ MockRaftActorContext context = createActorContext();
long term = 1000;
context.getTermInformation().update(term, null);
assertEquals("isVoteGranted", true, reply.isVoteGranted());
assertEquals("getTerm", term, reply.getTerm());
+ verify(follower).scheduleElection(any(FiniteDuration.class));
}
@Test
public void testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId(){
logStart("testHandleRequestVoteWhenSenderTermEqualToCurrentTermAndVotedForIsNotTheSameAsCandidateId");
- RaftActorContext context = createActorContext();
+ MockRaftActorContext context = createActorContext();
long term = 1000;
context.getTermInformation().update(term, "test");
RequestVoteReply reply = MessageCollectorActor.expectFirstMatching(leaderActor, RequestVoteReply.class);
assertEquals("isVoteGranted", false, reply.isVoteGranted());
+ verify(follower, never()).scheduleElection(any(FiniteDuration.class));
}
MockRaftActorContext context = createActorContext();
context.getReplicatedLog().clear(0,2);
- context.getReplicatedLog().append(newReplicatedLogEntry(1,100, "bar"));
+ context.getReplicatedLog().append(newReplicatedLogEntry(1, 100, "bar"));
context.getReplicatedLog().setSnapshotIndex(99);
List<ReplicatedLogEntry> entries = Arrays.asList(
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");
logStart("testHandleInstallSnapshot");
MockRaftActorContext context = createActorContext();
+ context.getTermInformation().update(1, "leader");
follower = createBehavior(context);
InstallSnapshot lastInstallSnapshot = null;
for(int i = 0; i < totalChunks; i++) {
- ByteString chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
+ byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
chunkData, chunkIndex, totalChunks);
follower.handleMessage(leaderActor, lastInstallSnapshot);
snapshot.getLastAppliedIndex());
assertEquals("getLastTerm", lastInstallSnapshot.getLastIncludedTerm(), snapshot.getLastTerm());
Assert.assertArrayEquals("getState", bsSnapshot.toByteArray(), snapshot.getState());
+ assertEquals("getElectionTerm", 1, snapshot.getElectionTerm());
+ assertEquals("getElectionVotedFor", "leader", snapshot.getElectionVotedFor());
+ applySnapshot.getCallback().onSuccess();
List<InstallSnapshotReply> replies = MessageCollectorActor.getAllMatching(
leaderActor, InstallSnapshotReply.class);
assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
}
- assertNull("Expected null SnapshotTracker", ((Follower) follower).getSnapshotTracker());
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
}
int lastIncludedIndex = 1;
// Check that snapshot installation is not in progress
- assertNull(((Follower) follower).getSnapshotTracker());
+ 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
- ByteString chunkData = getNextChunk(bsSnapshot, 0, chunkSize);
+ 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) follower).getSnapshotTracker());
+ assertNotNull(follower.getSnapshotTracker());
// Send an append entry
AppendEntries appendEntries = mock(AppendEntries.class);
InstallSnapshot lastInstallSnapshot = null;
for(int i = 0; i < totalChunks; i++) {
- ByteString chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
+ byte[] chunkData = getNextChunk(bsSnapshot, offset, chunkSize);
lastInstallSnapshot = new InstallSnapshot(1, "leader", lastIncludedIndex, 1,
chunkData, chunkIndex, totalChunks);
follower.handleMessage(leaderActor, lastInstallSnapshot);
assertEquals("getTerm", 1, reply.getTerm());
assertEquals("getFollowerId", context.getId(), reply.getFollowerId());
- assertNull("Expected null SnapshotTracker", ((Follower) follower).getSnapshotTracker());
+ assertNull("Expected null SnapshotTracker", follower.getSnapshotTracker());
}
@Test
MessageCollectorActor.assertNoneMatching(followerActor, ElectionTimeout.class, 500);
}
+ @Test
+ public void testElectionScheduledWhenAnyRaftRPCReceived(){
+ MockRaftActorContext context = createActorContext();
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, new RaftRPC() {
+ private static final long serialVersionUID = 1L;
- public ByteString getNextChunk (ByteString bs, int offset, int chunkSize){
+ @Override
+ public long getTerm() {
+ return 100;
+ }
+ });
+ verify(follower).scheduleElection(any(FiniteDuration.class));
+ }
+
+ @Test
+ public void testElectionNotScheduledWhenNonRaftRPCMessageReceived(){
+ MockRaftActorContext context = createActorContext();
+ follower = createBehavior(context);
+ follower.handleMessage(leaderActor, "non-raft-rpc");
+ verify(follower, never()).scheduleElection(any(FiniteDuration.class));
+ }
+
+ public byte[] getNextChunk (ByteString bs, int offset, int chunkSize){
int snapshotLength = bs.size();
int start = offset;
int size = chunkSize;
size = snapshotLength - start;
}
}
- return bs.substring(start, start + size);
+
+ 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("getLogLastTerm", expLogLastTerm, reply.getLogLastTerm());
assertEquals("getLogLastIndex", expLogLastIndex, reply.getLogLastIndex());
assertEquals("getPayloadVersion", payloadVersion, reply.getPayloadVersion());
+ assertEquals("isForceInstallSnapshot", expForceInstallSnapshot, reply.isForceInstallSnapshot());
}
- private ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
+
+ private static ReplicatedLogEntry newReplicatedLogEntry(long term, long index, String data) {
return new MockRaftActorContext.MockReplicatedLogEntry(term, index,
new MockRaftActorContext.MockPayload(data));
}
}
@Override
- protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(RaftActorContext actorContext,
+ protected void assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(MockRaftActorContext actorContext,
ActorRef actorRef, RaftRPC rpc) throws Exception {
super.assertStateChangesToFollowerWhenRaftRPCHasNewerTerm(actorContext, actorRef, rpc);
- String expVotedFor = RequestVote.class.isInstance(rpc) ? ((RequestVote)rpc).getCandidateId() : null;
+ String expVotedFor = rpc instanceof RequestVote ? ((RequestVote)rpc).getCandidateId() : null;
assertEquals("New votedFor", expVotedFor, actorContext.getTermInformation().getVotedFor());
}
@Override
- protected void handleAppendEntriesAddSameEntryToLogReply(TestActorRef<MessageCollectorActor> replyActor)
+ protected void handleAppendEntriesAddSameEntryToLogReply(final TestActorRef<MessageCollectorActor> replyActor)
throws Exception {
AppendEntriesReply reply = MessageCollectorActor.expectFirstMatching(replyActor, AppendEntriesReply.class);
assertEquals("isSuccess", true, reply.isSuccess());