- Refactored RaftActorBehavior#handleMessage (and related methods) to return RaftActorBehavior
instead of RaftActorState.
- Moved behavior switching from RaftActor based on requested state to the implementations of the
behaviors that should control that switch (Leader/Follower/Candidate) based on the type of
message received.
- Pull up logger reference for concrete RAFT behaviors to parent for a common LOG reference.
- Updated all logger calls to:
- Use substitution instead of String concatenation.
- Debug logger calls are gated by isDebugEnabled check.
- Correctly use Akka logging API for stack trace on exception.
- Avoid toString() calls during logging.
- Use assigned logger instance (LOG) instead of call to context.getLogger()
Change-Id: I376e21734a0540aae714fe9ef1562d74e73e558b
Signed-off-by: mark.mozolewski <mark.mozolewski@hp.com>
try {
bs = fromObject(state);
} catch (Exception e) {
- LOG.error("Exception in creating snapshot", e);
+ LOG.error(e, "Exception in creating snapshot");
}
getSelf().tell(new CaptureSnapshotReply(bs), null);
}
try {
state.putAll((HashMap) toObject(snapshot));
} catch (Exception e) {
- LOG.error("Exception in applying snapshot", e);
+ LOG.error(e, "Exception in applying snapshot");
}
if(LOG.isDebugEnabled()) {
- LOG.debug("Snapshot applied to state :" + ((HashMap) state).size());
+ LOG.debug("Snapshot applied to state : {}", ((HashMap) state).size());
}
}
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
-import org.opendaylight.controller.cluster.raft.behaviors.Candidate;
import org.opendaylight.controller.cluster.raft.behaviors.Follower;
-import org.opendaylight.controller.cluster.raft.behaviors.Leader;
import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
import org.opendaylight.controller.cluster.raft.client.messages.AddRaftPeer;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
}
private void onRecoveredSnapshot(SnapshotOffer offer) {
- LOG.debug("SnapshotOffer called..");
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("SnapshotOffer called..");
+ }
initRecoveryTimer();
replicatedLog.lastIndex(), replicatedLog.snapshotIndex,
replicatedLog.snapshotTerm, replicatedLog.size());
- currentBehavior = switchBehavior(RaftState.Follower);
+ currentBehavior = new Follower(context);
onStateChanged();
}
if (!(message instanceof AppendEntriesMessages.AppendEntries)
&& !(message instanceof AppendEntriesReply) && !(message instanceof SendHeartBeat)) {
if(LOG.isDebugEnabled()) {
- LOG.debug("onReceiveCommand: message:" + message.getClass());
+ LOG.debug("onReceiveCommand: message: {}", message.getClass());
}
}
- RaftState state =
- currentBehavior.handleMessage(getSender(), message);
RaftActorBehavior oldBehavior = currentBehavior;
- currentBehavior = switchBehavior(state);
+ currentBehavior = currentBehavior.handleMessage(getSender(), message);
+
if(oldBehavior != currentBehavior){
onStateChanged();
}
protected void onLeaderChanged(String oldLeader, String newLeader){};
- private RaftActorBehavior switchBehavior(RaftState state) {
- if (currentBehavior != null) {
- if (currentBehavior.state() == state) {
- return currentBehavior;
- }
- LOG.info("Switching from state " + currentBehavior.state() + " to "
- + state);
-
- try {
- currentBehavior.close();
- } catch (Exception e) {
- LOG.error(e,
- "Failed to close behavior : " + currentBehavior.state());
- }
-
- } else {
- LOG.info("Switching behavior to " + state);
- }
- RaftActorBehavior behavior = null;
- if (state == RaftState.Candidate) {
- behavior = new Candidate(context);
- } else if (state == RaftState.Follower) {
- behavior = new Follower(context);
- } else {
- behavior = new Leader(context);
- }
-
-
-
- return behavior;
- }
-
private void trimPersistentData(long sequenceNumber) {
// Trim akka snapshots
// FIXME : Not sure how exactly the SnapshotSelectionCriteria is applied
}
String peerAddress = context.getPeerAddress(leaderId);
if(LOG.isDebugEnabled()) {
- LOG.debug("getLeaderAddress leaderId = " + leaderId + " peerAddress = "
- + peerAddress);
+ LOG.debug("getLeaderAddress leaderId = {} peerAddress = {}",
+ leaderId, peerAddress);
}
return peerAddress;
public void appendAndPersist(final ActorRef clientActor,
final String identifier,
final ReplicatedLogEntry replicatedLogEntry) {
- context.getLogger().debug(
- "Append log entry and persist {} ", replicatedLogEntry);
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Append log entry and persist {} ", replicatedLogEntry);
+ }
+
// FIXME : By adding the replicated log entry to the in-memory journal we are not truly ensuring durability of the logs
journal.add(replicatedLogEntry);
import akka.actor.ActorRef;
import akka.actor.Cancellable;
+import akka.event.LoggingAdapter;
import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
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.SerializationUtils;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyLogEntries;
*/
protected final RaftActorContext context;
+ /**
+ *
+ */
+ protected final LoggingAdapter LOG;
+
/**
*
*/
protected AbstractRaftActorBehavior(RaftActorContext context) {
this.context = context;
+ this.LOG = context.getLogger();
}
/**
* @param appendEntries The AppendEntries message
* @return
*/
- protected abstract RaftState handleAppendEntries(ActorRef sender,
+ protected abstract RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries);
* @param appendEntries
* @return
*/
- protected RaftState appendEntries(ActorRef sender,
+ protected RaftActorBehavior appendEntries(ActorRef sender,
AppendEntries appendEntries) {
// 1. Reply false if term < currentTerm (ยง5.1)
if (appendEntries.getTerm() < currentTerm()) {
- context.getLogger().debug(
- "Cannot append entries because sender term " + appendEntries
- .getTerm() + " is less than " + currentTerm());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Cannot append entries because sender term {} is less than {}",
+ appendEntries.getTerm(), currentTerm());
+ }
+
sender.tell(
new AppendEntriesReply(context.getId(), currentTerm(), false,
lastIndex(), lastTerm()), actor()
);
- return state();
+ return this;
}
* @param appendEntriesReply The AppendEntriesReply message
* @return
*/
- protected abstract RaftState handleAppendEntriesReply(ActorRef sender,
+ protected abstract RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
AppendEntriesReply appendEntriesReply);
/**
* @param requestVote
* @return
*/
- protected RaftState requestVote(ActorRef sender,
+ protected RaftActorBehavior requestVote(ActorRef sender,
RequestVote requestVote) {
-
- context.getLogger().debug(requestVote.toString());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(requestVote.toString());
+ }
boolean grantVote = false;
sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor());
- return state();
+ return this;
}
/**
* @param requestVoteReply The RequestVoteReply message
* @return
*/
- protected abstract RaftState handleRequestVoteReply(ActorRef sender,
+ protected abstract RaftActorBehavior handleRequestVoteReply(ActorRef sender,
RequestVoteReply requestVoteReply);
/**
} else {
//if one index is not present in the log, no point in looping
// around as the rest wont be present either
- context.getLogger().warning(
- "Missing index {} from log. Cannot apply state. Ignoring {} to {}", i, i, index );
+ LOG.warning(
+ "Missing index {} from log. Cannot apply state. Ignoring {} to {}", i, i, index);
break;
}
}
- context.getLogger().debug("Setting last applied to {}", newLastApplied);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Setting last applied to {}", newLastApplied);
+ }
context.setLastApplied(newLastApplied);
// send a message to persist a ApplyLogEntries marker message into akka's persistent journal
}
@Override
- public RaftState handleMessage(ActorRef sender, Object message) {
+ public RaftActorBehavior handleMessage(ActorRef sender, Object message) {
if (message instanceof AppendEntries) {
return appendEntries(sender, (AppendEntries) message);
} else if (message instanceof AppendEntriesReply) {
} else if (message instanceof RequestVoteReply) {
return handleRequestVoteReply(sender, (RequestVoteReply) message);
}
- return state();
+ return this;
}
@Override public String getLeaderId() {
return leaderId;
}
+
+ protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) {
+ LOG.info("Switching from behavior {} to {}", this.state(), behavior.state());
+ try {
+ close();
+ } catch (Exception e) {
+ LOG.error(e, "Failed to close behavior : {}", this.state());
+ }
+
+ return behavior;
+ }
}
peers = context.getPeerAddresses().keySet();
- context.getLogger().debug("Election:Candidate has following peers:"+ peers);
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Election:Candidate has following peers: {}", peers);
+ }
if(peers.size() > 0) {
// Votes are required from a majority of the peers including self.
scheduleElection(electionDuration());
}
- @Override protected RaftState handleAppendEntries(ActorRef sender,
+ @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries) {
- context.getLogger().debug(appendEntries.toString());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug(appendEntries.toString());
+ }
- return state();
+ return this;
}
- @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
+ @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
AppendEntriesReply appendEntriesReply) {
- return state();
+ return this;
}
- @Override protected RaftState handleRequestVoteReply(ActorRef sender,
+ @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
RequestVoteReply requestVoteReply) {
if (requestVoteReply.isVoteGranted()) {
}
if (voteCount >= votesRequired) {
- return RaftState.Leader;
+ return switchBehavior(new Leader(context));
}
- return state();
+ return this;
}
@Override public RaftState state() {
}
@Override
- public RaftState handleMessage(ActorRef sender, Object originalMessage) {
+ public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
Object message = fromSerializableMessage(originalMessage);
RaftRPC rpc = (RaftRPC) message;
- context.getLogger().debug("RaftRPC message received {} my term is {}", rpc.toString(), context.getTermInformation().getCurrentTerm());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("RaftRPC message received {} my term is {}", rpc, context.getTermInformation().getCurrentTerm());
+ }
// If RPC request or response contains term T > currentTerm:
// set currentTerm = T, convert to follower (ยง5.1)
// This applies to all RPC messages and responses
if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
- return RaftState.Follower;
+
+ return switchBehavior(new Follower(context));
}
}
// ourselves the leader. This gives enough time for a leader
// who we do not know about (as a peer)
// to send a message to the candidate
- return RaftState.Leader;
+
+ return switchBehavior(new Leader(context));
}
startNewTerm();
scheduleElection(electionDuration());
- return state();
+ return this;
}
return super.handleMessage(sender, message);
context.getTermInformation().updateAndPersist(currentTerm + 1,
context.getId());
- context.getLogger().debug("Starting new term " + (currentTerm + 1));
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Starting new term {}", (currentTerm + 1));
+ }
// Request for a vote
// TODO: Retry request for vote if replies do not arrive in a reasonable
package org.opendaylight.controller.cluster.raft.behaviors;
import akka.actor.ActorRef;
-import akka.event.LoggingAdapter;
import com.google.protobuf.ByteString;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.opendaylight.controller.cluster.raft.RaftState;
public class Follower extends AbstractRaftActorBehavior {
private ByteString snapshotChunksCollected = ByteString.EMPTY;
- private final LoggingAdapter LOG;
-
public Follower(RaftActorContext context) {
super(context);
- LOG = context.getLogger();
-
scheduleElection(electionDuration());
}
- @Override protected RaftState handleAppendEntries(ActorRef sender,
+ @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries) {
if(appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) {
new AppendEntriesReply(context.getId(), currentTerm(), false,
lastIndex(), lastTerm()), actor()
);
- return state();
+ return this;
}
if (appendEntries.getEntries() != null
&& appendEntries.getEntries().size() > 0) {
if(LOG.isDebugEnabled()) {
LOG.debug(
- "Number of entries to be appended = " + appendEntries
- .getEntries().size()
+ "Number of entries to be appended = {}", appendEntries.getEntries().size()
);
}
if(LOG.isDebugEnabled()) {
LOG.debug(
- "Removing entries from log starting at "
- + matchEntry.getIndex()
+ "Removing entries from log starting at {}", matchEntry.getIndex()
);
}
}
if(LOG.isDebugEnabled()) {
- context.getLogger().debug(
- "After cleanup entries to be added from = " + (addEntriesFrom
- + lastIndex())
+ LOG.debug("After cleanup entries to be added from = {}", (addEntriesFrom + lastIndex())
);
}
for (int i = addEntriesFrom;
i < appendEntries.getEntries().size(); i++) {
- context.getLogger().info(
- "Append entry to log " + appendEntries.getEntries().get(
- i).getData()
- .toString()
- );
- context.getReplicatedLog()
- .appendAndPersist(appendEntries.getEntries().get(i));
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Append entry to log {}", appendEntries.getEntries().get(i).getData());
+ }
+ context.getReplicatedLog().appendAndPersist(appendEntries.getEntries().get(i));
}
if(LOG.isDebugEnabled()) {
- LOG.debug("Log size is now " + context.getReplicatedLog().size());
+ LOG.debug("Log size is now {}", context.getReplicatedLog().size());
}
}
if (prevCommitIndex != context.getCommitIndex()) {
if(LOG.isDebugEnabled()) {
- LOG.debug("Commit index set to " + context.getCommitIndex());
+ LOG.debug("Commit index set to {}", context.getCommitIndex());
}
}
sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true,
lastIndex(), lastTerm()), actor());
- return state();
+ return this;
}
- @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
+ @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
AppendEntriesReply appendEntriesReply) {
- return state();
+ return this;
}
- @Override protected RaftState handleRequestVoteReply(ActorRef sender,
+ @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
RequestVoteReply requestVoteReply) {
- return state();
+ return this;
}
@Override public RaftState state() {
return RaftState.Follower;
}
- @Override public RaftState handleMessage(ActorRef sender, Object originalMessage) {
+ @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
Object message = fromSerializableMessage(originalMessage);
}
if (message instanceof ElectionTimeout) {
- return RaftState.Candidate;
+ return switchBehavior(new Candidate(context));
} else if (message instanceof InstallSnapshot) {
InstallSnapshot installSnapshot = (InstallSnapshot) message;
// this is the last chunk, create a snapshot object and apply
snapshotChunksCollected = snapshotChunksCollected.concat(installSnapshot.getData());
- context.getLogger().debug("Last chunk received: snapshotChunksCollected.size:{}",
- snapshotChunksCollected.size());
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Last chunk received: snapshotChunksCollected.size:{}",
+ snapshotChunksCollected.size());
+ }
Snapshot snapshot = Snapshot.create(snapshotChunksCollected.toByteArray(),
new ArrayList<ReplicatedLogEntry>(),
true), actor());
} catch (Exception e) {
- context.getLogger().error("Exception in InstallSnapshot of follower", e);
+ LOG.error(e, "Exception in InstallSnapshot of follower:");
//send reply with success as false. The chunk will be sent again on failure
sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
installSnapshot.getChunkIndex(), false), actor());
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.Cancellable;
-import akka.event.LoggingAdapter;
import com.google.common.base.Preconditions;
import com.google.protobuf.ByteString;
import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
private final int minReplicationCount;
- private final LoggingAdapter LOG;
-
public Leader(RaftActorContext context) {
super(context);
- LOG = context.getLogger();
-
followers = context.getPeerAddresses().keySet();
for (String followerId : followers) {
}
if(LOG.isDebugEnabled()) {
- LOG.debug("Election:Leader has following peers:" + followers);
+ LOG.debug("Election:Leader has following peers: {}", followers);
}
if (followers.size() > 0) {
}
- @Override protected RaftState handleAppendEntries(ActorRef sender,
+ @Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries) {
if(LOG.isDebugEnabled()) {
LOG.debug(appendEntries.toString());
}
- return state();
+ return this;
}
- @Override protected RaftState handleAppendEntriesReply(ActorRef sender,
+ @Override protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
AppendEntriesReply appendEntriesReply) {
if(! appendEntriesReply.isSuccess()) {
if(followerLogInformation == null){
LOG.error("Unknown follower {}", followerId);
- return state();
+ return this;
}
if (appendEntriesReply.isSuccess()) {
applyLogToStateMachine(context.getCommitIndex());
}
- return state();
+ return this;
}
protected ClientRequestTracker removeClientRequestTracker(long logIndex) {
return null;
}
- @Override protected RaftState handleRequestVoteReply(ActorRef sender,
+ @Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
RequestVoteReply requestVoteReply) {
- return state();
+ return this;
}
@Override public RaftState state() {
return RaftState.Leader;
}
- @Override public RaftState handleMessage(ActorRef sender, Object originalMessage) {
+ @Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
Preconditions.checkNotNull(sender, "sender should not be null");
Object message = fromSerializableMessage(originalMessage);
// This applies to all RPC messages and responses
if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
- return RaftState.Follower;
+
+ return switchBehavior(new Follower(context));
}
}
try {
if (message instanceof SendHeartBeat) {
- return sendHeartBeat();
+ sendHeartBeat();
+ return this;
} else if(message instanceof SendInstallSnapshot) {
installSnapshotIfNeeded();
} else if (message instanceof Replicate) {
long logIndex = replicate.getReplicatedLogEntry().getIndex();
if(LOG.isDebugEnabled()) {
- LOG.debug("Replicate message " + logIndex);
+ LOG.debug("Replicate message {}", logIndex);
}
// Create a tracker entry we will use this later to notify the
followerActor.path(), mapFollowerToSnapshot.get(followerId).getChunkIndex(),
mapFollowerToSnapshot.get(followerId).getTotalChunks());
} catch (IOException e) {
- LOG.error("InstallSnapshot failed for Leader.", e);
+ LOG.error(e, "InstallSnapshot failed for Leader.");
}
}
return nextChunk;
}
- private RaftState sendHeartBeat() {
+ private void sendHeartBeat() {
if (followers.size() > 0) {
sendAppendEntries();
}
- return state();
}
private void stopHeartBeat() {
* differently.
*/
public interface RaftActorBehavior extends AutoCloseable{
+
/**
* Handle a message. If the processing of the message warrants a state
- * change then a new state should be returned otherwise this method should
- * return the state for the current behavior.
+ * change then a new behavior should be returned otherwise this method should
+ * return the current behavior.
*
* @param sender The sender of the message
* @param message A message that needs to be processed
*
- * @return The new state or self (this)
+ * @return The new behavior or current behavior
*/
- RaftState handleMessage(ActorRef sender, Object message);
+ RaftActorBehavior handleMessage(ActorRef sender, Object message);
/**
* The state associated with a given behavior
return true;
}
}.from(raftActor.path().toString())
- .message("Switching from state Candidate to Leader")
+ .message("Switching from behavior Candidate to Leader")
.occurrences(1).exec();
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.SerializationUtils;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import java.util.List;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
public abstract class AbstractRaftActorBehaviorTest extends AbstractActorTest {
RaftActorBehavior behavior = createBehavior(context);
// Send an unknown message so that the state of the RaftActor remains unchanged
- RaftState expected = behavior.handleMessage(getRef(), "unknown");
+ RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
behavior.handleMessage(getRef(), appendEntries);
- assertEquals(expected, raftState);
+ assertEquals(expected, raftBehavior);
// Also expect an AppendEntriesReply to be sent where success is false
final Boolean out = new ExpectMsg<Boolean>(duration("1 seconds"),
}
// Send an unknown message so that the state of the RaftActor remains unchanged
- RaftState expected = behavior.handleMessage(getRef(), "unknown");
+ RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
behavior.handleMessage(getRef(), appendEntries);
- assertEquals(expected, raftState);
+ assertEquals(expected, raftBehavior);
assertEquals(1, log.size());
RaftActorBehavior behavior = createBehavior(
createActorContext(behaviorActor));
- RaftState raftState = behavior.handleMessage(getTestActor(),
+ RaftActorBehavior raftBehavior = behavior.handleMessage(getTestActor(),
new RequestVote(1000, "test", 10000, 999));
- if(behavior.state() != RaftState.Follower){
- assertEquals(RaftState.Follower, raftState);
+ if(!(behavior instanceof Follower)){
+ assertTrue(raftBehavior instanceof Follower);
} else {
final Boolean out =
RaftActorBehavior behavior = createBehavior(actorContext);
- RaftState raftState = behavior.handleMessage(getTestActor(),
+ RaftActorBehavior raftBehavior = behavior.handleMessage(getTestActor(),
new RequestVote(1000, "test", 10000, 999));
- if(behavior.state() != RaftState.Follower){
- assertEquals(RaftState.Follower, raftState);
+ if(!(behavior instanceof Follower)){
+ assertTrue(raftBehavior instanceof Follower);
} else {
final Boolean out =
new ExpectMsg<Boolean>(duration("1 seconds"),
setLastLogEntry(
(MockRaftActorContext) actorContext, 0, 0, p);
- RaftState raftState = createBehavior(actorContext)
+ RaftActorBehavior raftBehavior = createBehavior(actorContext)
.handleMessage(actorRef, rpc);
- assertEquals(RaftState.Follower, raftState);
+ assertTrue(raftBehavior instanceof Follower);
}
protected MockRaftActorContext.SimpleReplicatedLog setLastLogEntry(
import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
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.base.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
Candidate candidate =
new Candidate(raftActorContext);
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
candidate.handleMessage(candidateActor, new ElectionTimeout());
- Assert.assertEquals(RaftState.Leader, raftState);
+ Assert.assertTrue(raftBehavior instanceof Leader);
}
@Test
Candidate candidate =
new Candidate(raftActorContext);
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
candidate.handleMessage(candidateActor, new ElectionTimeout());
- Assert.assertEquals(RaftState.Candidate, raftState);
+ Assert.assertTrue(raftBehavior instanceof Candidate);
}
@Test
Candidate candidate =
new Candidate(raftActorContext);
- RaftState stateOnFirstVote = candidate.handleMessage(peerActor1, new RequestVoteReply(0, true));
+ RaftActorBehavior behaviorOnFirstVote = candidate.handleMessage(peerActor1, new RequestVoteReply(0, true));
- Assert.assertEquals(RaftState.Leader, stateOnFirstVote);
+ Assert.assertTrue(behaviorOnFirstVote instanceof Leader);
}
Candidate candidate =
new Candidate(raftActorContext);
- RaftState stateOnFirstVote = candidate.handleMessage(peerActor1, new RequestVoteReply(0, true));
+ RaftActorBehavior behaviorOnFirstVote = candidate.handleMessage(peerActor1, new RequestVoteReply(0, true));
- RaftState stateOnSecondVote = candidate.handleMessage(peerActor2, new RequestVoteReply(0, true));
+ RaftActorBehavior behaviorOnSecondVote = candidate.handleMessage(peerActor2, new RequestVoteReply(0, true));
- Assert.assertEquals(RaftState.Candidate, stateOnFirstVote);
- Assert.assertEquals(RaftState.Leader, stateOnSecondVote);
+ Assert.assertTrue(behaviorOnFirstVote instanceof Candidate);
+ Assert.assertTrue(behaviorOnSecondVote instanceof Leader);
}
import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
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.base.messages.ApplySnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
Follower follower =
new Follower(raftActorContext);
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
follower.handleMessage(followerActor, new ElectionTimeout());
- Assert.assertEquals(RaftState.Candidate, raftState);
+ Assert.assertTrue(raftBehavior instanceof Candidate);
}
@Test
AppendEntries appendEntries =
new AppendEntries(2, "leader-1", 100, 1, entries, 101);
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
createBehavior(context).handleMessage(getRef(), appendEntries);
assertEquals(101L, context.getLastApplied());
RaftActorBehavior behavior = createBehavior(context);
// Send an unknown message so that the state of the RaftActor remains unchanged
- RaftState expected = behavior.handleMessage(getRef(), "unknown");
+ RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
behavior.handleMessage(getRef(), appendEntries);
- assertEquals(expected, raftState);
+ assertEquals(expected, raftBehavior);
// Also expect an AppendEntriesReply to be sent where success is false
final Boolean out = new ExpectMsg<Boolean>(duration("1 seconds"),
RaftActorBehavior behavior = createBehavior(context);
// Send an unknown message so that the state of the RaftActor remains unchanged
- RaftState expected = behavior.handleMessage(getRef(), "unknown");
+ RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
behavior.handleMessage(getRef(), appendEntries);
- assertEquals(expected, raftState);
+ assertEquals(expected, raftBehavior);
assertEquals(5, log.last().getIndex() + 1);
assertNotNull(log.get(3));
assertNotNull(log.get(4));
RaftActorBehavior behavior = createBehavior(context);
// Send an unknown message so that the state of the RaftActor remains unchanged
- RaftState expected = behavior.handleMessage(getRef(), "unknown");
+ RaftActorBehavior expected = behavior.handleMessage(getRef(), "unknown");
- RaftState raftState =
+ RaftActorBehavior raftBehavior =
behavior.handleMessage(getRef(), appendEntries);
- assertEquals(expected, raftState);
+ assertEquals(expected, raftBehavior);
// The entry at index 2 will be found out-of-sync with the leader
// and will be removed
import org.opendaylight.controller.cluster.raft.FollowerLogInformationImpl;
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.ReplicatedLogImplEntry;
import org.opendaylight.controller.cluster.raft.SerializationUtils;
import org.opendaylight.controller.cluster.raft.base.messages.ApplyState;
// handle message should return the Leader state when it receives an
// unknown message
- RaftState state = leader.handleMessage(senderActor, "foo");
- Assert.assertEquals(RaftState.Leader, state);
+ RaftActorBehavior behavior = leader.handleMessage(senderActor, "foo");
+ Assert.assertTrue(behavior instanceof Leader);
}};
}
actorContext.setPeerAddresses(peerAddresses);
Leader leader = new Leader(actorContext);
- RaftState raftState = leader
+ RaftActorBehavior raftBehavior = leader
.handleMessage(senderActor, new Replicate(null, null,
new MockRaftActorContext.MockReplicatedLogEntry(1,
100,
));
// State should not change
- assertEquals(RaftState.Leader, raftState);
+ assertTrue(raftBehavior instanceof Leader);
final String out =
new ExpectMsg<String>(duration("1 seconds"), "match hint") {
.build());
Leader leader = new Leader(actorContext);
- RaftState raftState = leader
+ RaftActorBehavior raftBehavior = leader
.handleMessage(senderActor, new Replicate(null, "state-id",actorContext.getReplicatedLog().get(1)));
// State should not change
- assertEquals(RaftState.Leader, raftState);
+ assertTrue(raftBehavior instanceof Leader);
assertEquals(1, actorContext.getCommitIndex());
new MockRaftActorContext.MockPayload("D"));
// this should invoke a sendinstallsnapshot as followersLastIndex < snapshotIndex
- RaftState raftState = leader.handleMessage(
+ RaftActorBehavior raftBehavior = leader.handleMessage(
senderActor, new Replicate(null, "state-id", entry));
- assertEquals(RaftState.Leader, raftState);
+ assertTrue(raftBehavior instanceof Leader);
// we might receive some heartbeat messages, so wait till we SendInstallSnapshot
Boolean[] matches = new ReceiveWhile<Boolean>(Boolean.class, duration("2 seconds")) {
new ReplicatedLogImplEntry(newEntryIndex, currentTerm,
new MockRaftActorContext.MockPayload("D"));
- RaftState raftState = leader.handleMessage(senderActor, new SendInstallSnapshot());
+ RaftActorBehavior raftBehavior = leader.handleMessage(senderActor, new SendInstallSnapshot());
- assertEquals(RaftState.Leader, raftState);
+ assertTrue(raftBehavior instanceof Leader);
// check if installsnapshot gets called with the correct values.
final String out =
//clears leaders log
actorContext.getReplicatedLog().removeFrom(0);
- RaftState raftState = leader.handleMessage(senderActor,
+ RaftActorBehavior raftBehavior = leader.handleMessage(senderActor,
new InstallSnapshotReply(currentTerm, followerActor.path().toString(),
leader.getFollowerToSnapshot().getChunkIndex(), true));
- assertEquals(RaftState.Leader, raftState);
+ assertTrue(raftBehavior instanceof Leader);
assertEquals(leader.mapFollowerToSnapshot.size(), 0);
assertEquals(leader.followerToLog.size(), 1);
Code Review / controller.git / commitdiff