import akka.cluster.MemberStatus;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Stopwatch;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
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.ApplyState;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.base.messages.TimeoutNow;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
* convert to candidate
* </ul>
*/
+// Non-final for testing
public class Follower extends AbstractRaftActorBehavior {
private static final long MAX_ELECTION_TIMEOUT_FACTOR = 18;
this(context, null, (short)-1);
}
+ @SuppressFBWarnings(value = "MC_OVERRIDABLE_METHOD_CALL_IN_CONSTRUCTOR",
+ justification = "electionDuration() is not final for Candidate override")
public Follower(final RaftActorContext context, final String initialLeaderId,
final short initialLeaderPayloadVersion) {
super(context, RaftState.Follower);
- this.leaderId = initialLeaderId;
- this.leaderPayloadVersion = initialLeaderPayloadVersion;
+ leaderId = initialLeaderId;
+ leaderPayloadVersion = initialLeaderPayloadVersion;
initialSyncStatusTracker = new SyncStatusTracker(context.getActor(), getId(), context.getConfigParams()
.getSyncIndexThreshold());
return this;
}
+ @Override
+ final ApplyState getApplyStateFor(final ReplicatedLogEntry entry) {
+ return new ApplyState(null, null, entry);
+ }
+
@Override
public RaftActorBehavior handleMessage(final ActorRef sender, final Object message) {
if (message instanceof ElectionTimeout || message instanceof TimeoutNow) {
// 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()) {
+ if (rpc.getTerm() > context.getTermInformation().getCurrentTerm() && shouldUpdateTerm(rpc)) {
log.info("{}: Term {} in \"{}\" message is greater than follower's term {} - updating term",
logName(), rpc.getTerm(), rpc, context.getTermInformation().getCurrentTerm());
if (isLeaderAvailabilityKnown() && lastLeaderMessageInterval < maxElectionTimeout) {
log.debug("{}: Received ElectionTimeout but leader appears to be available", logName());
scheduleElection(electionDuration());
+ } else if (isThisFollowerIsolated()) {
+ log.debug("{}: this follower is isolated. Do not switch to Candidate for now.", logName());
+ setLeaderId(null);
+ scheduleElection(electionDuration());
} else {
log.debug("{}: Received ElectionTimeout - switching to Candidate", logName());
return internalSwitchBehavior(RaftState.Candidate);
return false;
}
+ private boolean isThisFollowerIsolated() {
+ final Optional<Cluster> maybeCluster = context.getCluster();
+ if (!maybeCluster.isPresent()) {
+ return false;
+ }
+
+ final Cluster cluster = maybeCluster.get();
+ final Member selfMember = cluster.selfMember();
+
+ final CurrentClusterState state = cluster.state();
+ final Set<Member> unreachable = state.getUnreachable();
+ final Iterable<Member> members = state.getMembers();
+
+ log.debug("{}: Checking if this node is isolated in the cluster unreachable set {},"
+ + "all members {} self member: {}", logName(), unreachable, members, selfMember);
+
+ // no unreachable peers means we cannot be isolated
+ if (unreachable.size() == 0) {
+ return false;
+ }
+
+ final Set<Member> membersToCheck = new HashSet<>();
+ members.forEach(membersToCheck::add);
+
+ membersToCheck.removeAll(unreachable);
+
+ // check if the only member not unreachable is us
+ if (membersToCheck.size() == 1 && membersToCheck.iterator().next().equals(selfMember)) {
+ return true;
+ }
+
+ return false;
+ }
+
private void handleInstallSnapshot(final ActorRef sender, final InstallSnapshot installSnapshot) {
log.debug("{}: handleInstallSnapshot: {}", logName(), installSnapshot);