import akka.actor.ActorRef;
import akka.actor.Cancellable;
import com.google.common.base.Preconditions;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.cluster.raft.ClientRequestTracker;
/**
* Used for message logging.
*/
+ @SuppressFBWarnings("SLF4J_LOGGER_SHOULD_BE_PRIVATE")
protected final Logger log;
/**
}
@Override
- public void setReplicatedToAllIndex(long replicatedToAllIndex) {
+ public void setReplicatedToAllIndex(final long replicatedToAllIndex) {
this.replicatedToAllIndex = replicatedToAllIndex;
}
* @param appendEntries the message
* @return a new behavior if it was changed or the current behavior
*/
- protected RaftActorBehavior appendEntries(ActorRef sender, AppendEntries appendEntries) {
+ protected RaftActorBehavior appendEntries(final ActorRef sender, final AppendEntries appendEntries) {
// 1. Reply false if term < currentTerm (§5.1)
if (appendEntries.getTerm() < currentTerm()) {
- log.debug("{}: Cannot append entries because sender term {} is less than {}", logName(),
+ log.info("{}: Cannot append entries because sender's term {} is less than {}", logName(),
appendEntries.getTerm(), currentTerm());
sender.tell(new AppendEntriesReply(context.getId(), currentTerm(), false, lastIndex(), lastTerm(),
- context.getPayloadVersion()), actor());
+ context.getPayloadVersion(), false, false, appendEntries.getLeaderRaftVersion()), actor());
return this;
}
* @param requestVote the message
* @return a new behavior if it was changed or the current behavior
*/
- protected RaftActorBehavior requestVote(ActorRef sender, RequestVote requestVote) {
+ protected RaftActorBehavior requestVote(final ActorRef sender, final RequestVote requestVote) {
- log.debug("{}: In requestVote: {}", logName(), requestVote);
+ log.debug("{}: In requestVote: {} - currentTerm: {}, votedFor: {}, lastIndex: {}, lastTerm: {}", logName(),
+ requestVote, currentTerm(), votedFor(), lastIndex(), lastTerm());
boolean grantVote = canGrantVote(requestVote);
return this;
}
- protected boolean canGrantVote(RequestVote requestVote) {
+ protected boolean canGrantVote(final RequestVote requestVote) {
boolean grantVote = false;
// Reply false if term < currentTerm (§5.1)
*
* @param interval the duration after which we should trigger a new election
*/
- protected void scheduleElection(FiniteDuration interval) {
+ protected void scheduleElection(final FiniteDuration interval) {
stopElection();
// Schedule an election. When the scheduler triggers an ElectionTimeout message is sent to itself
* @param logIndex the log index
* @return the ClientRequestTracker or null if none available
*/
- protected ClientRequestTracker removeClientRequestTracker(long logIndex) {
+ protected ClientRequestTracker removeClientRequestTracker(final long logIndex) {
return null;
}
*
* @return the log entry index or -1 if not found
*/
- protected long getLogEntryIndex(long index) {
+ protected long getLogEntryIndex(final long index) {
if (index == context.getReplicatedLog().getSnapshotIndex()) {
return context.getReplicatedLog().getSnapshotIndex();
}
}
/**
- * Returns the actual term of the entry in replicated log for the given index or -1 if not found.
+ * Returns the actual term of the entry in the replicated log for the given index or -1 if not found.
*
* @return the log entry term or -1 if not found
*/
- protected long getLogEntryTerm(long index) {
+ protected long getLogEntryTerm(final long index) {
if (index == context.getReplicatedLog().getSnapshotIndex()) {
return context.getReplicatedLog().getSnapshotTerm();
}
return -1;
}
+ /**
+ * Returns the actual term of the entry in the replicated log for the given index or, if not present, returns the
+ * snapshot term if the given index is in the snapshot or -1 otherwise.
+ *
+ * @return the term or -1 otherwise
+ */
+ protected long getLogEntryOrSnapshotTerm(final long index) {
+ if (context.getReplicatedLog().isInSnapshot(index)) {
+ return context.getReplicatedLog().getSnapshotTerm();
+ }
+
+ return getLogEntryTerm(index);
+ }
+
/**
* Applies the log entries up to the specified index that is known to be committed to the state machine.
*
* @param index the log index
*/
protected void applyLogToStateMachine(final long index) {
- long newLastApplied = context.getLastApplied();
// Now maybe we apply to the state machine
for (long i = context.getLastApplied() + 1; i < index + 1; i++) {
// Send a local message to the local RaftActor (it's derived class to be
// specific to apply the log to it's index)
- final ApplyState msg;
+ final ApplyState applyState;
final ClientRequestTracker tracker = removeClientRequestTracker(i);
if (tracker != null) {
- msg = new ApplyState(tracker.getClientActor(), tracker.getIdentifier(), replicatedLogEntry);
+ applyState = new ApplyState(tracker.getClientActor(), tracker.getIdentifier(), replicatedLogEntry);
} else {
- msg = new ApplyState(null, null, replicatedLogEntry);
+ applyState = new ApplyState(null, null, replicatedLogEntry);
}
- actor().tell(msg, actor());
- newLastApplied = i;
+ log.debug("{}: Setting last applied to {}", logName(), i);
+
+ context.setLastApplied(i);
+ context.getApplyStateConsumer().accept(applyState);
} else {
//if one index is not present in the log, no point in looping
// around as the rest wont be present either
}
}
- log.debug("{}: Setting last applied to {}", logName(), newLastApplied);
-
- context.setLastApplied(newLastApplied);
-
// send a message to persist a ApplyLogEntries marker message into akka's persistent journal
// will be used during recovery
//in case if the above code throws an error and this message is not sent, it would be fine
}
@Override
- public RaftActorBehavior handleMessage(ActorRef sender, Object message) {
+ public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) {
if (message instanceof AppendEntries) {
return appendEntries(sender, (AppendEntries) message);
} else if (message instanceof AppendEntriesReply) {
}
@Override
- public RaftActorBehavior switchBehavior(RaftActorBehavior behavior) {
+ public RaftActorBehavior switchBehavior(final RaftActorBehavior behavior) {
return internalSwitchBehavior(behavior);
}
- protected RaftActorBehavior internalSwitchBehavior(RaftState newState) {
+ protected RaftActorBehavior internalSwitchBehavior(final RaftState newState) {
return internalSwitchBehavior(createBehavior(context, newState));
}
@SuppressWarnings("checkstyle:IllegalCatch")
- protected RaftActorBehavior internalSwitchBehavior(RaftActorBehavior newBehavior) {
+ protected RaftActorBehavior internalSwitchBehavior(final RaftActorBehavior newBehavior) {
if (!context.getRaftPolicy().automaticElectionsEnabled()) {
return this;
}
- log.info("{} :- Switching from behavior {} to {}", logName(), this.state(), newBehavior.state());
+ log.info("{} :- Switching from behavior {} to {}, election term: {}", logName(), this.state(),
+ newBehavior.state(), context.getTermInformation().getCurrentTerm());
try {
close();
} catch (RuntimeException e) {
}
- protected int getMajorityVoteCount(int numPeers) {
+ protected int getMajorityVoteCount(final int numPeers) {
// Votes are required from a majority of the peers including self.
// The numMajority field therefore stores a calculated value
// of the number of votes required for this candidate to win an
Code Review / controller.git / blobdiff