* 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 java.util.Objects.requireNonNull;
+
import akka.actor.ActorRef;
import akka.actor.ActorSelection;
import akka.actor.Cancellable;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
import com.google.common.io.ByteSource;
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.IOException;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
-import javax.annotation.Nullable;
+import org.eclipse.jdt.annotation.Nullable;
import org.opendaylight.controller.cluster.io.SharedFileBackedOutputStream;
import org.opendaylight.controller.cluster.messaging.MessageSlicer;
import org.opendaylight.controller.cluster.messaging.SliceOptions;
private int minReplicationCount;
protected AbstractLeader(final RaftActorContext context, final RaftState state,
- @Nullable final AbstractLeader initializeFromLeader) {
+ final @Nullable AbstractLeader initializeFromLeader) {
super(context, state);
appendEntriesMessageSlicer = MessageSlicer.builder().logContext(logName())
}
@VisibleForTesting
- @SuppressFBWarnings(value = "NP_PARAMETER_MUST_BE_NONNULL_BUT_MARKED_AS_NULLABLE",
- justification = "Unrecognised NullableDecl")
- void setSnapshotHolder(@Nullable final SnapshotHolder snapshotHolder) {
+ void setSnapshotHolder(final @Nullable SnapshotHolder snapshotHolder) {
this.snapshotHolder = Optional.fromNullable(snapshotHolder);
}
followerLogInformation.markFollowerActive();
followerLogInformation.setPayloadVersion(appendEntriesReply.getPayloadVersion());
followerLogInformation.setRaftVersion(appendEntriesReply.getRaftVersion());
+ followerLogInformation.setNeedsLeaderAddress(appendEntriesReply.isNeedsLeaderAddress());
long followerLastLogIndex = appendEntriesReply.getLogLastIndex();
- long followersLastLogTermInLeadersLog = getLogEntryTerm(followerLastLogIndex);
boolean updated = false;
if (appendEntriesReply.getLogLastIndex() > context.getReplicatedLog().lastIndex()) {
// The follower's log is actually ahead of the leader's log. Normally this doesn't happen
// However in this case the log terms won't match and the logs will conflict - this is handled
// elsewhere.
log.info("{}: handleAppendEntriesReply: follower {} lastIndex {} is ahead of our lastIndex {} "
- + "(snapshotIndex {}) - forcing install snaphot", logName(), followerLogInformation.getId(),
- appendEntriesReply.getLogLastIndex(), context.getReplicatedLog().lastIndex(),
- context.getReplicatedLog().getSnapshotIndex());
+ + "(snapshotIndex {}, snapshotTerm {}) - forcing install snaphot", logName(),
+ followerLogInformation.getId(), appendEntriesReply.getLogLastIndex(),
+ context.getReplicatedLog().lastIndex(), context.getReplicatedLog().getSnapshotIndex(),
+ context.getReplicatedLog().getSnapshotTerm());
followerLogInformation.setMatchIndex(-1);
followerLogInformation.setNextIndex(-1);
updated = true;
} else if (appendEntriesReply.isSuccess()) {
+ long followersLastLogTermInLeadersLog = getLogEntryTerm(followerLastLogIndex);
if (followerLastLogIndex >= 0 && followersLastLogTermInLeadersLog >= 0
&& followersLastLogTermInLeadersLog != appendEntriesReply.getLogLastTerm()) {
// The follower's last entry is present in the leader's journal but the terms don't match so the
updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply);
}
} else {
- log.info("{}: handleAppendEntriesReply - received unsuccessful reply: {}, leader snapshotIndex: {}",
- logName(), appendEntriesReply, context.getReplicatedLog().getSnapshotIndex());
+ log.info("{}: handleAppendEntriesReply - received unsuccessful reply: {}, leader snapshotIndex: {}, "
+ + "snapshotTerm: {}, replicatedToAllIndex: {}", logName(), appendEntriesReply,
+ context.getReplicatedLog().getSnapshotIndex(), context.getReplicatedLog().getSnapshotTerm(),
+ getReplicatedToAllIndex());
+ long followersLastLogTermInLeadersLogOrSnapshot = getLogEntryOrSnapshotTerm(followerLastLogIndex);
if (appendEntriesReply.isForceInstallSnapshot()) {
// Reset the followers match and next index. This is to signal that this follower has nothing
// in common with this Leader and so would require a snapshot to be installed
// Force initiate a snapshot capture
initiateCaptureSnapshot(followerId);
- } else if (followerLastLogIndex < 0 || followersLastLogTermInLeadersLog >= 0
- && followersLastLogTermInLeadersLog == appendEntriesReply.getLogLastTerm()) {
- // The follower's log is empty or the last entry is present in the leader's journal
- // and the terms match so the follower is just behind the leader's journal from
- // the last snapshot, if any. We'll catch up the follower quickly by starting at the
- // follower's last log index.
+ } else if (followerLastLogIndex < 0 || followersLastLogTermInLeadersLogOrSnapshot >= 0
+ && followersLastLogTermInLeadersLogOrSnapshot == appendEntriesReply.getLogLastTerm()) {
+ // The follower's log is empty or the follower's last entry is present in the leader's journal or
+ // snapshot and the terms match so the follower is just behind the leader's journal from the last
+ // snapshot, if any. We'll catch up the follower quickly by starting at the follower's last log index.
updated = updateFollowerLogInformation(followerLogInformation, appendEntriesReply);
+ "updated: matchIndex: {}, nextIndex: {}", logName(), followerId,
followerLogInformation.getMatchIndex(), followerLogInformation.getNextIndex());
} else {
- // The follower's log conflicts with leader's log so decrement follower's next index by 1
+ // The follower's log conflicts with leader's log so decrement follower's next index
// in an attempt to find where the logs match.
-
- if (followerLogInformation.decrNextIndex()) {
+ if (followerLogInformation.decrNextIndex(appendEntriesReply.getLogLastIndex())) {
updated = true;
log.info("{}: follower {} last log term {} conflicts with the leader's {} - dec next index to {}",
logName(), followerId, appendEntriesReply.getLogLastTerm(),
- followersLastLogTermInLeadersLog, followerLogInformation.getNextIndex());
+ followersLastLogTermInLeadersLogOrSnapshot, followerLogInformation.getNextIndex());
}
}
}
@Override
public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) {
- Preconditions.checkNotNull(sender, "sender should not be null");
+ requireNonNull(sender, "sender should not be null");
if (appendEntriesMessageSlicer.handleMessage(message)) {
return this;
return this;
}
+ @SuppressFBWarnings(value = "NP_NULL_PARAM_DEREF_ALL_TARGETS_DANGEROUS",
+ justification = "JDT nullness with SpotBugs at setSnapshotHolder(null)")
private void handleInstallSnapshotReply(final InstallSnapshotReply reply) {
log.debug("{}: handleInstallSnapshotReply: {}", logName(), reply);
AppendEntries appendEntries = new AppendEntries(currentTerm(), context.getId(),
getLogEntryIndex(followerNextIndex - 1),
getLogEntryTerm(followerNextIndex - 1), entries,
- leaderCommitIndex, super.getReplicatedToAllIndex(), context.getPayloadVersion());
+ leaderCommitIndex, super.getReplicatedToAllIndex(), context.getPayloadVersion(),
+ followerLogInformation.getRaftVersion(), followerLogInformation.needsLeaderAddress(getId()));
if (!entries.isEmpty() || log.isTraceEnabled()) {
log.debug("{}: Sending AppendEntries to follower {}: {}", logName(), followerLogInformation.getId(),
Code Review / controller.git / blobdiff