Added more logging in the raft behavior classes where appropriate.
Reduced logging in some places to trace or removed redundant logging to
avoid too much output when debug level is enabled.
Removed isDebugEnabled() checks in various places where not needed to
improve test line coverage.
Also I found it useful to see the behavior's state in the log messages.
So I added a logName field and method to AbstractRaftBehavior that
is printed in each log message instead of just context.getId().
Also added/modified toString() in a few RaftRPC message classes.
Change-Id: Ic4d3c93e25d5caf16319b8433d54cab69b26f5c0
Signed-off-by: tpantelis <tpanteli@brocade.com>
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.CaptureSnapshotReply;
import org.opendaylight.controller.cluster.raft.base.messages.Replicate;
-import org.opendaylight.controller.cluster.raft.base.messages.SendHeartBeat;
import org.opendaylight.controller.cluster.raft.base.messages.SendInstallSnapshot;
import org.opendaylight.controller.cluster.raft.behaviors.AbstractRaftActorBehavior;
import org.opendaylight.controller.cluster.raft.behaviors.Follower;
import org.opendaylight.controller.cluster.raft.behaviors.RaftActorBehavior;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeader;
import org.opendaylight.controller.cluster.raft.client.messages.FindLeaderReply;
-import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
import org.opendaylight.controller.cluster.raft.protobuff.client.messages.Payload;
-import org.opendaylight.controller.protobuff.messages.cluster.raft.AppendEntriesMessages;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
handleCaptureSnapshotReply(((CaptureSnapshotReply) message).getSnapshot());
} else {
- if (!(message instanceof AppendEntriesMessages.AppendEntries)
- && !(message instanceof AppendEntriesReply) && !(message instanceof SendHeartBeat)) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: onReceiveCommand: message: {}", persistenceId(), message.getClass());
- }
- }
-
RaftActorBehavior oldBehavior = currentBehavior;
currentBehavior = currentBehavior.handleMessage(getSender(), message);
private Optional<ByteString> snapshot;
public AbstractLeader(RaftActorContext context) {
- super(context);
+ super(context, RaftState.Leader);
final Builder<String, FollowerLogInformation> ftlBuilder = ImmutableMap.builder();
for (String followerId : context.getPeerAddresses().keySet()) {
leaderId = context.getId();
- LOG.debug("{}: Election: Leader has following peers: {}", context.getId(), getFollowerIds());
+ LOG.debug("{}: Election: Leader has following peers: {}", logName(), getFollowerIds());
minReplicationCount = getMajorityVoteCount(getFollowerIds().size());
// Upon election: send initial empty AppendEntries RPCs
// (heartbeat) to each server; repeat during idle periods to
// prevent election timeouts (§5.2)
- sendAppendEntries(0);
+ sendAppendEntries(0, false);
}
/**
return followerToLog.keySet();
}
- private Optional<ByteString> getSnapshot() {
- return snapshot;
- }
-
@VisibleForTesting
void setSnapshot(Optional<ByteString> snapshot) {
this.snapshot = snapshot;
protected RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
- }
+ LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
return this;
}
protected RaftActorBehavior handleAppendEntriesReply(ActorRef sender,
AppendEntriesReply appendEntriesReply) {
- if(! appendEntriesReply.isSuccess()) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntriesReply: {}", context.getId(), appendEntriesReply);
- }
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply);
+ } else if(LOG.isDebugEnabled() && !appendEntriesReply.isSuccess()) {
+ LOG.debug("{}: handleAppendEntriesReply: {}", logName(), appendEntriesReply);
}
// Update the FollowerLogInformation
followerToLog.get(followerId);
if(followerLogInformation == null){
- LOG.error("{}: handleAppendEntriesReply - unknown follower {}", context.getId(), followerId);
+ LOG.error("{}: handleAppendEntriesReply - unknown follower {}", logName(), followerId);
return this;
}
// Apply the change to the state machine
if (context.getCommitIndex() > context.getLastApplied()) {
+ LOG.debug("{}: handleAppendEntriesReply: applying to log - commitIndex: {}, lastAppliedIndex: {}",
+ logName(), context.getCommitIndex(), context.getLastApplied());
+
applyLogToStateMachine(context.getCommitIndex());
}
}
//Send the next log entry immediately, if possible, no need to wait for heartbeat to trigger that event
- sendUpdatesToFollower(followerId, followerLogInformation, false);
+ sendUpdatesToFollower(followerId, followerLogInformation, false, false);
return this;
}
return this;
}
- @Override
- public RaftState state() {
- return RaftState.Leader;
- }
-
@Override
public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
Preconditions.checkNotNull(sender, "sender should not be null");
// set currentTerm = T, convert to follower (§5.1)
// This applies to all RPC messages and responses
if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
- LOG.debug("{}: Term {} in \"{}\" message is greater than leader's term {}", context.getId(),
- rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
+ LOG.debug("{}: Term {} in \"{}\" message is greater than leader's term {} - switching to Follower",
+ logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
}
private void handleInstallSnapshotReply(InstallSnapshotReply reply) {
+ LOG.debug("{}: handleInstallSnapshotReply: {}", logName(), reply);
+
String followerId = reply.getFollowerId();
FollowerToSnapshot followerToSnapshot = mapFollowerToSnapshot.get(followerId);
if (followerToSnapshot == null) {
LOG.error("{}: FollowerId {} in InstallSnapshotReply not known to Leader",
- context.getId(), followerId);
+ logName(), followerId);
return;
}
//this was the last chunk reply
if(LOG.isDebugEnabled()) {
LOG.debug("{}: InstallSnapshotReply received, " +
- "last chunk received, Chunk:{}. Follower:{} Setting nextIndex:{}",
- context.getId(), reply.getChunkIndex(), followerId,
+ "last chunk received, Chunk: {}. Follower: {} Setting nextIndex: {}",
+ logName(), reply.getChunkIndex(), followerId,
context.getReplicatedLog().getSnapshotIndex() + 1
);
}
context.getReplicatedLog().getSnapshotIndex() + 1);
mapFollowerToSnapshot.remove(followerId);
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: followerToLog.get(followerId).getNextIndex()=" +
- context.getId(), followerToLog.get(followerId).getNextIndex());
- }
+ LOG.debug("{}: follower: {}, matchIndex set to {}, nextIndex set to {}",
+ logName(), followerId, followerLogInformation.getMatchIndex(),
+ followerLogInformation.getNextIndex());
if (mapFollowerToSnapshot.isEmpty()) {
// once there are no pending followers receiving snapshots
}
} else {
LOG.info("{}: InstallSnapshotReply received sending snapshot chunk failed, Will retry, Chunk: {}",
- context.getId(), reply.getChunkIndex());
+ logName(), reply.getChunkIndex());
followerToSnapshot.markSendStatus(false);
}
} else {
LOG.error("{}: Chunk index {} in InstallSnapshotReply from follower {} does not match expected index {}",
- context.getId(), reply.getChunkIndex(), followerId,
+ logName(), reply.getChunkIndex(), followerId,
followerToSnapshot.getChunkIndex());
if(reply.getChunkIndex() == INVALID_CHUNK_INDEX){
private void replicate(Replicate replicate) {
long logIndex = replicate.getReplicatedLogEntry().getIndex();
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Replicate message {}", context.getId(), logIndex);
- }
+ LOG.debug("{}: Replicate message: identifier: {}, logIndex: {}", logName(),
+ replicate.getIdentifier(), logIndex);
// Create a tracker entry we will use this later to notify the
// client actor
context.setCommitIndex(logIndex);
applyLogToStateMachine(logIndex);
} else {
- sendAppendEntries(0);
+ sendAppendEntries(0, false);
}
}
- private void sendAppendEntries(long timeSinceLastActivityInterval) {
+ private void sendAppendEntries(long timeSinceLastActivityInterval, boolean isHeartbeat) {
// Send an AppendEntries to all followers
for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
final String followerId = e.getKey();
// This checks helps not to send a repeat message to the follower
if(!followerLogInformation.isFollowerActive() ||
followerLogInformation.timeSinceLastActivity() >= timeSinceLastActivityInterval) {
- sendUpdatesToFollower(followerId, followerLogInformation, true);
+ sendUpdatesToFollower(followerId, followerLogInformation, true, isHeartbeat);
}
}
}
*/
private void sendUpdatesToFollower(String followerId, FollowerLogInformation followerLogInformation,
- boolean sendHeartbeat) {
+ boolean sendHeartbeat, boolean isHeartbeat) {
ActorSelection followerActor = context.getPeerActorSelection(followerId);
if (followerActor != null) {
} else {
long leaderLastIndex = context.getReplicatedLog().lastIndex();
long leaderSnapShotIndex = context.getReplicatedLog().getSnapshotIndex();
- if (isFollowerActive &&
- context.getReplicatedLog().isPresent(followerNextIndex)) {
+
+ if(!isHeartbeat || LOG.isTraceEnabled()) {
+ LOG.debug("{}: Checking sendAppendEntries for follower {}, leaderLastIndex: {}, leaderSnapShotIndex: {}",
+ logName(), followerId, leaderLastIndex, leaderSnapShotIndex);
+ }
+
+ if (isFollowerActive && context.getReplicatedLog().isPresent(followerNextIndex)) {
+
+ LOG.debug("{}: sendAppendEntries: {} is present for follower {}", logName(),
+ followerNextIndex, followerId);
+
// FIXME : Sending one entry at a time
final List<ReplicatedLogEntry> entries = context.getReplicatedLog().getFrom(followerNextIndex, 1);
// then snapshot should be sent
if (LOG.isDebugEnabled()) {
- LOG.debug("InitiateInstallSnapshot to follower:{}," +
- "follower-nextIndex:{}, leader-snapshot-index:{}, " +
- "leader-last-index:{}", followerId,
- followerNextIndex, leaderSnapShotIndex, leaderLastIndex
- );
+ LOG.debug(String.format("%s: InitiateInstallSnapshot to follower: %s," +
+ "follower-nextIndex: %d, leader-snapshot-index: %d, " +
+ "leader-last-index: %d", logName(), followerId,
+ followerNextIndex, leaderSnapShotIndex, leaderLastIndex));
}
// Send heartbeat to follower whenever install snapshot is initiated.
prevLogTerm(followerNextIndex), entries,
context.getCommitIndex(), super.getReplicatedToAllIndex());
- if(!entries.isEmpty()) {
- LOG.debug("{}: Sending AppendEntries to follower {}: {}", context.getId(), followerId,
+ if(!entries.isEmpty() || LOG.isTraceEnabled()) {
+ LOG.debug("{}: Sending AppendEntries to follower {}: {}", logName(), followerId,
appendEntries);
}
} else if (!context.isSnapshotCaptureInitiated()) {
- LOG.info("{}: Initiating Snapshot Capture to Install Snapshot, Leader:{}", context.getId(), getLeaderId());
+ LOG.info("{}: Initiating Snapshot Capture to Install Snapshot, Leader:{}", logName(), getLeaderId());
ReplicatedLogEntry lastAppliedEntry = context.getReplicatedLog().get(context.getLastApplied());
long lastAppliedIndex = -1;
long lastAppliedTerm = -1;
private void sendInstallSnapshot() {
+ LOG.debug("{}: sendInstallSnapshot", logName());
for (Entry<String, FollowerLogInformation> e : followerToLog.entrySet()) {
ActorSelection followerActor = context.getPeerActorSelection(e.getKey());
actor()
);
LOG.info("{}: InstallSnapshot sent to follower {}, Chunk: {}/{}",
- context.getId(), followerActor.path(),
+ logName(), followerActor.path(),
followerToSnapshot.getChunkIndex(),
followerToSnapshot.getTotalChunks());
}
} catch (IOException e) {
- LOG.error("{}: InstallSnapshot failed for Leader.", context.getId(), e);
+ LOG.error("{}: InstallSnapshot failed for Leader.", logName(), e);
}
}
mapFollowerToSnapshot.put(followerId, followerToSnapshot);
}
ByteString nextChunk = followerToSnapshot.getNextChunk();
- if (LOG.isDebugEnabled()) {
- LOG.debug("{}: Leader's snapshot nextChunk size:{}", context.getId(), nextChunk.size());
- }
+
+ LOG.debug("{}: next snapshot chunk size for follower {}: {}", logName(), followerId, nextChunk.size());
+
return nextChunk;
}
private void sendHeartBeat() {
if (!followerToLog.isEmpty()) {
- sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toMillis());
+ LOG.trace("{}: Sending heartbeat", logName());
+ sendAppendEntries(context.getConfigParams().getHeartBeatInterval().toMillis(), true);
}
}
((size % context.getConfigParams().getSnapshotChunkSize()) > 0 ? 1 : 0);
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Snapshot {} bytes, total chunks to send:{}",
- context.getId(), size, totalChunks);
+ logName(), size, totalChunks);
}
replyReceivedForOffset = -1;
chunkIndex = AbstractLeader.FIRST_CHUNK_INDEX;
}
}
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Next chunk: length={}, offset={},size={}", context.getId(),
+
+ LOG.debug("{}: Next chunk: length={}, offset={},size={}", logName(),
snapshotLength, start, size);
- }
+
ByteString substring = getSnapshotBytes().substring(start, start + size);
nextChunkHashCode = substring.hashCode();
return substring;
import java.util.concurrent.TimeUnit;
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;
private long replicatedToAllIndex = -1;
- protected AbstractRaftActorBehavior(RaftActorContext context) {
+ private final String logName;
+
+ private final RaftState state;
+
+ protected AbstractRaftActorBehavior(RaftActorContext context, RaftState state) {
this.context = context;
+ this.state = state;
this.LOG = context.getLogger();
+
+ logName = String.format("%s (%s)", context.getId(), state);
+ }
+
+ @Override
+ public RaftState state() {
+ return state;
+ }
+
+ public String logName() {
+ return logName;
}
@Override
if (appendEntries.getTerm() < currentTerm()) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: Cannot append entries because sender term {} is less than {}",
- context.getId(), appendEntries.getTerm(), currentTerm());
+ logName(), appendEntries.getTerm(), currentTerm());
}
sender.tell(
* @param requestVote
* @return
*/
- protected RaftActorBehavior requestVote(ActorRef sender,
- RequestVote requestVote) {
+ protected RaftActorBehavior requestVote(ActorRef sender, RequestVote requestVote) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Received {}", context.getId(), requestVote);
- }
+ LOG.debug("{}: In requestVote: {}", logName(), requestVote);
boolean grantVote = false;
}
}
- sender.tell(new RequestVoteReply(currentTerm(), grantVote), actor());
+ RequestVoteReply reply = new RequestVoteReply(currentTerm(), grantVote);
+
+ LOG.debug("{}: requestVote returning: {}", logName(), reply);
+
+ sender.tell(reply, actor());
return this;
}
// around as the rest wont be present either
LOG.warn(
"{}: Missing index {} from log. Cannot apply state. Ignoring {} to {}",
- context.getId(), i, i, index);
+ logName(), i, i, index);
break;
}
}
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Setting last applied to {}", context.getId(), newLastApplied);
+ LOG.debug("{}: Setting last applied to {}", logName(), newLastApplied);
}
context.setLastApplied(newLastApplied);
}
protected RaftActorBehavior switchBehavior(RaftActorBehavior behavior) {
- LOG.info("{} :- Switching from behavior {} to {}", context.getId(), this.state(), behavior.state());
+ LOG.info("{} :- Switching from behavior {} to {}", logName(), this.state(), behavior.state());
try {
close();
} catch (Exception e) {
- LOG.error("{}: Failed to close behavior : {}", context.getId(), this.state(), e);
+ LOG.error("{}: Failed to close behavior : {}", logName(), this.state(), e);
}
return behavior;
long tempMin = Math.min(snapshotCapturedIndex, (lastApplied > -1 ? lastApplied - 1 : -1));
if (tempMin > -1 && context.getReplicatedLog().isPresent(tempMin)) {
+ LOG.debug("{}: fakeSnapshot purging log to {} for term {}", logName(), tempMin,
+ context.getTermInformation().getCurrentTerm());
+
//use the term of the temp-min, since we check for isPresent, entry will not be null
ReplicatedLogEntry entry = context.getReplicatedLog().get(tempMin);
context.getReplicatedLog().snapshotPreCommit(tempMin, entry.getTerm());
private final Set<String> peers;
public Candidate(RaftActorContext context) {
- super(context);
+ super(context, RaftState.Candidate);
peers = context.getPeerAddresses().keySet();
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Election: Candidate has following peers: {}", context.getId(), peers);
+ LOG.debug("{}: Election: Candidate has following peers: {}", logName(), peers);
}
votesRequired = getMajorityVoteCount(peers.size());
AppendEntries appendEntries) {
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
+ LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
}
return this;
}
@Override protected RaftActorBehavior handleRequestVoteReply(ActorRef sender,
- RequestVoteReply requestVoteReply) {
+ RequestVoteReply requestVoteReply) {
+
+ LOG.debug("{}: handleRequestVoteReply: {}, current voteCount: {}", logName(), requestVoteReply,
+ voteCount);
if (requestVoteReply.isVoteGranted()) {
voteCount++;
return this;
}
- @Override public RaftState state() {
- return RaftState.Candidate;
- }
-
@Override
public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
RaftRPC rpc = (RaftRPC) message;
if(LOG.isDebugEnabled()) {
- LOG.debug("{}: RaftRPC message received {} my term is {}", context.getId(), rpc,
+ LOG.debug("{}: RaftRPC message received {}, my term is {}", logName(), rpc,
context.getTermInformation().getCurrentTerm());
}
}
if (message instanceof ElectionTimeout) {
+ LOG.debug("{}: Received ElectionTimeout", logName());
+
if (votesRequired == 0) {
// If there are no peers then we should be a Leader
// We wait for the election timeout to occur before declare
// Increment the election term and vote for self
long currentTerm = context.getTermInformation().getCurrentTerm();
- context.getTermInformation().updateAndPersist(currentTerm + 1,
- context.getId());
+ long newTerm = currentTerm + 1;
+ context.getTermInformation().updateAndPersist(newTerm, context.getId());
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Starting new term {}", context.getId(), (currentTerm + 1));
- }
+ LOG.debug("{}: Starting new term {}", logName(), newTerm);
// Request for a vote
// TODO: Retry request for vote if replies do not arrive in a reasonable
for (String peerId : peers) {
ActorSelection peerActor = context.getPeerActorSelection(peerId);
if(peerActor != null) {
- peerActor.tell(new RequestVote(
+ RequestVote requestVote = new RequestVote(
context.getTermInformation().getCurrentTerm(),
context.getId(),
context.getReplicatedLog().lastIndex(),
- context.getReplicatedLog().lastTerm()),
- context.getActor()
- );
- }
- }
+ context.getReplicatedLog().lastTerm());
+ LOG.debug("{}: Sending {} to peer {}", logName(), requestVote, peerId);
+ peerActor.tell(requestVote, context.getActor());
+ }
+ }
}
@Override public void close() throws Exception {
private SnapshotTracker snapshotTracker = null;
public Follower(RaftActorContext context) {
- super(context);
+ super(context, RaftState.Follower);
scheduleElection(electionDuration());
}
@Override protected RaftActorBehavior handleAppendEntries(ActorRef sender,
AppendEntries appendEntries) {
- if(appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: handleAppendEntries: {}", context.getId(), appendEntries);
- }
+ int numLogEntries = appendEntries.getEntries() != null ? appendEntries.getEntries().size() : 0;
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("{}: handleAppendEntries: {}", logName(), appendEntries);
+ } else if(LOG.isDebugEnabled() && numLogEntries > 0) {
+ LOG.debug("{}: handleAppendEntries: {}", logName(), appendEntries);
}
// TODO : Refactor this method into a bunch of smaller methods
boolean outOfSync = true;
// First check if the logs are in sync or not
- if (lastIndex() == -1
- && appendEntries.getPrevLogIndex() != -1) {
+ long lastIndex = lastIndex();
+ if (lastIndex == -1 && appendEntries.getPrevLogIndex() != -1) {
// The follower's log is out of sync because the leader does have
// an entry at prevLogIndex and this follower has no entries in
// it's log.
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: The followers log is empty and the senders prevLogIndex is {}",
- context.getId(), appendEntries.getPrevLogIndex());
- }
-
- } else if (lastIndex() > -1
- && appendEntries.getPrevLogIndex() != -1
- && !prevEntryPresent) {
+ LOG.debug("{}: The followers log is empty and the senders prevLogIndex is {}",
+ logName(), appendEntries.getPrevLogIndex());
+ } else if (lastIndex > -1 && appendEntries.getPrevLogIndex() != -1 && !prevEntryPresent) {
// The follower's log is out of sync because the Leader's
// prevLogIndex entry was not found in it's log
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it",
- context.getId(), appendEntries.getPrevLogIndex());
- }
-
- } else if (lastIndex() > -1
- && prevEntryPresent
- && prevLogTerm != appendEntries.getPrevLogTerm()) {
+ LOG.debug("{}: The log is not empty but the prevLogIndex {} was not found in it",
+ logName(), appendEntries.getPrevLogIndex());
+ } else if (lastIndex > -1 && prevEntryPresent && prevLogTerm != appendEntries.getPrevLogTerm()) {
// The follower's log is out of sync because the Leader's
// prevLogIndex entry does exist in the follower's log but it has
// a different term in it
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- "{}: Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}"
- , context.getId(), prevLogTerm
- , appendEntries.getPrevLogTerm());
- }
+ LOG.debug(
+ "{}: Cannot append entries because previous entry term {} is not equal to append entries prevLogTerm {}",
+ logName(), prevLogTerm, appendEntries.getPrevLogTerm());
} else {
outOfSync = false;
}
if (outOfSync) {
// We found that the log was out of sync so just send a negative
// reply and return
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Follower ({}) is out-of-sync, " +
- "so sending negative reply, lastIndex():{}, lastTerm():{}",
- context.getId(), context.getId(), lastIndex(), lastTerm()
- );
- }
- sender.tell(
- new AppendEntriesReply(context.getId(), currentTerm(), false,
- lastIndex(), lastTerm()), actor()
- );
+
+ LOG.debug("{}: Follower is out-of-sync, so sending negative reply, lastIndex: {}, lastTerm: {}",
+ logName(), lastIndex, lastTerm());
+
+ sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex,
+ lastTerm()), actor());
return this;
}
- if (appendEntries.getEntries() != null
- && appendEntries.getEntries().size() > 0) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Number of entries to be appended = {}", context.getId(),
+ if (appendEntries.getEntries() != null && appendEntries.getEntries().size() > 0) {
+
+ LOG.debug("{}: Number of entries to be appended = {}", logName(),
appendEntries.getEntries().size());
- }
// 3. If an existing entry conflicts with a new one (same index
// but different terms), delete the existing entry and all that
break;
}
- if (newEntry.getTerm() == matchEntry
- .getTerm()) {
+ if (newEntry.getTerm() == matchEntry.getTerm()) {
continue;
}
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Removing entries from log starting at {}", context.getId(),
+ LOG.debug("{}: Removing entries from log starting at {}", logName(),
matchEntry.getIndex());
- }
// Entries do not match so remove all subsequent entries
- context.getReplicatedLog()
- .removeFromAndPersist(matchEntry.getIndex());
+ context.getReplicatedLog().removeFromAndPersist(matchEntry.getIndex());
break;
}
}
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: After cleanup entries to be added from = {}", context.getId(),
- (addEntriesFrom + lastIndex()));
- }
+ lastIndex = lastIndex();
+ LOG.debug("{}: After cleanup entries to be added from = {}", logName(),
+ (addEntriesFrom + lastIndex));
// 4. Append any new entries not already in the log
- for (int i = addEntriesFrom;
- i < appendEntries.getEntries().size(); i++) {
+ for (int i = addEntriesFrom; i < appendEntries.getEntries().size(); i++) {
+ ReplicatedLogEntry entry = appendEntries.getEntries().get(i);
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Append entry to log {}", context.getId(),
- appendEntries.getEntries().get(i).getData());
- }
- context.getReplicatedLog().appendAndPersist(appendEntries.getEntries().get(i));
- }
+ LOG.debug("{}: Append entry to log {}", logName(), entry.getData());
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Log size is now {}", context.getId(), context.getReplicatedLog().size());
+ context.getReplicatedLog().appendAndPersist(entry);
}
- }
+ LOG.debug("{}: Log size is now {}", logName(), context.getReplicatedLog().size());
+ }
// 5. If leaderCommit > commitIndex, set commitIndex =
// min(leaderCommit, index of last new entry)
+ lastIndex = lastIndex();
long prevCommitIndex = context.getCommitIndex();
- context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(),
- context.getReplicatedLog().lastIndex()));
+ context.setCommitIndex(Math.min(appendEntries.getLeaderCommit(), lastIndex));
if (prevCommitIndex != context.getCommitIndex()) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: Commit index set to {}", context.getId(), context.getCommitIndex());
- }
+ LOG.debug("{}: Commit index set to {}", logName(), context.getCommitIndex());
}
// If commitIndex > lastApplied: increment lastApplied, apply
// log[lastApplied] to state machine (§5.3)
// check if there are any entries to be applied. last-applied can be equal to last-index
if (appendEntries.getLeaderCommit() > context.getLastApplied() &&
- context.getLastApplied() < lastIndex()) {
+ context.getLastApplied() < lastIndex) {
if(LOG.isDebugEnabled()) {
LOG.debug("{}: applyLogToStateMachine, " +
- "appendEntries.getLeaderCommit():{}," +
- "context.getLastApplied():{}, lastIndex():{}", context.getId(),
- appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex()
- );
+ "appendEntries.getLeaderCommit(): {}," +
+ "context.getLastApplied(): {}, lastIndex(): {}", logName(),
+ appendEntries.getLeaderCommit(), context.getLastApplied(), lastIndex);
}
applyLogToStateMachine(appendEntries.getLeaderCommit());
}
- sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), true,
- lastIndex(), lastTerm()), actor());
+ AppendEntriesReply reply = new AppendEntriesReply(context.getId(), currentTerm(), true,
+ lastIndex, lastTerm());
+
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("{}: handleAppendEntries returning : {}", logName(), reply);
+ } else if(LOG.isDebugEnabled() && numLogEntries > 0) {
+ LOG.debug("{}: handleAppendEntries returning : {}", logName(), reply);
+ }
+
+ sender.tell(reply, actor());
if (!context.isSnapshotCaptureInitiated()) {
super.performSnapshotWithoutCapture(appendEntries.getReplicatedToAllIndex());
return this;
}
- @Override public RaftState state() {
- return RaftState.Follower;
- }
-
@Override public RaftActorBehavior handleMessage(ActorRef sender, Object originalMessage) {
Object message = fromSerializableMessage(originalMessage);
// set currentTerm = T, convert to follower (§5.1)
// This applies to all RPC messages and responses
if (rpc.getTerm() > context.getTermInformation().getCurrentTerm()) {
+ LOG.debug("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term",
+ logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
+
context.getTermInformation().updateAndPersist(rpc.getTerm(), null);
}
}
if (message instanceof ElectionTimeout) {
+ LOG.debug("{}: Received ElectionTimeout - switching to Candidate", logName());
return switchBehavior(new Candidate(context));
} else if (message instanceof InstallSnapshot) {
private void handleInstallSnapshot(ActorRef sender, InstallSnapshot installSnapshot) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: InstallSnapshot received by follower " +
- "datasize:{} , Chunk:{}/{}", context.getId(), installSnapshot.getData().size(),
- installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks()
- );
- }
+
+ LOG.debug("{}: InstallSnapshot received from leader {}, datasize: {} , Chunk: {}/{}",
+ logName(), installSnapshot.getLeaderId(), installSnapshot.getData().size(),
+ installSnapshot.getChunkIndex(), installSnapshot.getTotalChunks());
if(snapshotTracker == null){
snapshotTracker = new SnapshotTracker(LOG, installSnapshot.getTotalChunks());
}
- sender.tell(new InstallSnapshotReply(
- currentTerm(), context.getId(), installSnapshot.getChunkIndex(),
- true), actor());
+ InstallSnapshotReply reply = new InstallSnapshotReply(
+ currentTerm(), context.getId(), installSnapshot.getChunkIndex(), true);
+
+ LOG.debug("{}: handleInstallSnapshot returning: {}", logName(), reply);
+
+ sender.tell(reply, actor());
} catch (SnapshotTracker.InvalidChunkException e) {
+ LOG.debug("{}: Exception in InstallSnapshot of follower", logName(), e);
sender.tell(new InstallSnapshotReply(currentTerm(), context.getId(),
-1, false), actor());
snapshotTracker = null;
} catch (Exception e){
- LOG.error("{}: Exception in InstallSnapshot of follower", context.getId(), e);
+ LOG.error("{}: Exception in InstallSnapshot of follower", logName(), e);
+
//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());
ByteString getSnapshotChunksCollected(){
return snapshotTracker != null ? snapshotTracker.getCollectedChunks() : ByteString.EMPTY;
}
-
-
}
return replicatedToAllIndex;
}
+
@Override
public String toString() {
- final StringBuilder sb =
- new StringBuilder("AppendEntries{");
- sb.append("term=").append(getTerm());
- sb.append("leaderId='").append(leaderId).append('\'');
- sb.append(", prevLogIndex=").append(prevLogIndex);
- sb.append(", prevLogTerm=").append(prevLogTerm);
- sb.append(", entries=").append(entries);
- sb.append(", leaderCommit=").append(leaderCommit);
- sb.append(", replicatedToAllIndex=").append(replicatedToAllIndex);
- sb.append('}');
- return sb.toString();
+ StringBuilder builder = new StringBuilder();
+ builder.append("AppendEntries [term=").append(term).append(", leaderId=").append(leaderId)
+ .append(", prevLogIndex=").append(prevLogIndex).append(", prevLogTerm=").append(prevLogTerm)
+ .append(", entries=").append(entries).append(", leaderCommit=").append(leaderCommit)
+ .append(", replicatedToAllIndex=").append(replicatedToAllIndex).append("]");
+ return builder.toString();
}
public <T extends Object> Object toSerializable() {
return followerId;
}
- @Override public String toString() {
- final StringBuilder sb =
- new StringBuilder("AppendEntriesReply{");
- sb.append("term=").append(term);
- sb.append(", success=").append(success);
- sb.append(", logLastIndex=").append(logLastIndex);
- sb.append(", logLastTerm=").append(logLastTerm);
- sb.append(", followerId='").append(followerId).append('\'');
- sb.append('}');
- return sb.toString();
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("AppendEntriesReply [term=").append(term).append(", success=").append(success)
+ .append(", logLastIndex=").append(logLastIndex).append(", logLastTerm=").append(logLastTerm)
+ .append(", followerId=").append(followerId).append("]");
+ return builder.toString();
}
}
return installSnapshot;
}
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("InstallSnapshot [term=").append(term).append(", leaderId=").append(leaderId)
+ .append(", lastIncludedIndex=").append(lastIncludedIndex).append(", lastIncludedTerm=")
+ .append(lastIncludedTerm).append(", data=").append(data).append(", chunkIndex=").append(chunkIndex)
+ .append(", totalChunks=").append(totalChunks).append(", lastChunkHashCode=").append(lastChunkHashCode)
+ .append("]");
+ return builder.toString();
+ }
}
public boolean isSuccess() {
return success;
}
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("InstallSnapshotReply [term=").append(term).append(", followerId=").append(followerId)
+ .append(", chunkIndex=").append(chunkIndex).append(", success=").append(success).append("]");
+ return builder.toString();
+ }
}
this.lastLogTerm = lastLogTerm;
}
- @Override public String toString() {
- final StringBuilder sb =
- new StringBuilder("RequestVote{");
- sb.append("term='").append(getTerm()).append('\'');
- sb.append("candidateId='").append(candidateId).append('\'');
- sb.append(", lastLogIndex=").append(lastLogIndex);
- sb.append(", lastLogTerm=").append(lastLogTerm);
- sb.append('}');
- return sb.toString();
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("RequestVote [term=").append(term).append(", candidateId=").append(candidateId)
+ .append(", lastLogIndex=").append(lastLogIndex).append(", lastLogTerm=").append(lastLogTerm)
+ .append("]");
+ return builder.toString();
}
}
public boolean isVoteGranted() {
return voteGranted;
}
+
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append("RequestVoteReply [term=").append(term).append(", voteGranted=").append(voteGranted).append("]");
+ return builder.toString();
+ }
}
protected final Logger LOG = LoggerFactory.getLogger(getClass());
public AbstractUntypedPersistentActor() {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Actor created {}", getSelf());
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Actor created {}", getSelf());
}
getContext().
system().
@Override public void onReceiveCommand(Object message) throws Exception {
final String messageType = message.getClass().getSimpleName();
- if(LOG.isDebugEnabled()) {
- LOG.debug("Received message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Received message {}", messageType);
}
handleCommand(message);
- if(LOG.isDebugEnabled()) {
- LOG.debug("Done handling message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Done handling message {}", messageType);
}
}
@Override public void onReceiveRecover(Object message) throws Exception {
final String messageType = message.getClass().getSimpleName();
- if(LOG.isDebugEnabled()) {
- LOG.debug("Received message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Received message {}", messageType);
}
handleRecover(message);
- if(LOG.isDebugEnabled()) {
- LOG.debug("Done handling message {}", messageType);
+ if(LOG.isTraceEnabled()) {
+ LOG.trace("Done handling message {}", messageType);
}
}
@Override
public void onReceiveCommand(final Object message) throws Exception {
- if(LOG.isDebugEnabled()) {
- LOG.debug("{}: onReceiveCommand: Received message {} from {}", persistenceId(), message, getSender());
- }
-
if (message.getClass().equals(CreateTransaction.SERIALIZABLE_CLASS)) {
handleCreateTransaction(message);
} else if(message instanceof ForwardedReadyTransaction) {
Code Review / controller.git / commitdiff