+ private RaftActorBehavior handleElectionTimeout(final Object message) {
+ // If the message is ElectionTimeout, verify we haven't actually seen a message from the leader
+ // during the election timeout interval. It may that the election timer expired b/c this actor
+ // was busy and messages got delayed, in which case leader messages would be backed up in the
+ // queue but would be processed before the ElectionTimeout message and thus would restart the
+ // lastLeaderMessageTimer.
+ long lastLeaderMessageInterval = lastLeaderMessageTimer.elapsed(TimeUnit.MILLISECONDS);
+ long electionTimeoutInMillis = context.getConfigParams().getElectionTimeOutInterval().toMillis();
+ boolean noLeaderMessageReceived = !lastLeaderMessageTimer.isRunning()
+ || lastLeaderMessageInterval >= electionTimeoutInMillis;
+
+ if (canStartElection()) {
+ if (message instanceof TimeoutNow) {
+ log.debug("{}: Received TimeoutNow - switching to Candidate", logName());
+ return internalSwitchBehavior(RaftState.Candidate);
+ } else if (noLeaderMessageReceived) {
+ // Check the cluster state to see if the leader is known to be up before we go to Candidate.
+ // However if we haven't heard from the leader in a long time even though the cluster state
+ // indicates it's up then something is wrong - leader might be stuck indefinitely - so switch
+ // to Candidate,
+ long maxElectionTimeout = electionTimeoutInMillis * MAX_ELECTION_TIMEOUT_FACTOR;
+ 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);
+ }
+ } else {
+ log.debug("{}: Received ElectionTimeout but lastLeaderMessageInterval {} < election timeout {}",
+ logName(), lastLeaderMessageInterval, context.getConfigParams().getElectionTimeOutInterval());
+ scheduleElection(electionDuration());
+ }
+ } else if (message instanceof ElectionTimeout) {
+ if (noLeaderMessageReceived) {
+ setLeaderId(null);
+ }
+
+ scheduleElection(electionDuration());
+ }
+
+ return this;
+ }
+
+ private boolean isLeaderAvailabilityKnown() {
+ if (leaderId == null) {
+ return false;
+ }
+
+ Optional<Cluster> cluster = context.getCluster();
+ if (!cluster.isPresent()) {
+ return false;
+ }
+
+ ActorSelection leaderActor = context.getPeerActorSelection(leaderId);
+ if (leaderActor == null) {
+ return false;
+ }
+
+ Address leaderAddress = leaderActor.anchorPath().address();
+
+ CurrentClusterState state = cluster.get().state();
+ Set<Member> unreachable = state.getUnreachable();
+
+ log.debug("{}: Checking for leader {} in the cluster unreachable set {}", logName(), leaderAddress,
+ unreachable);
+
+ for (Member m: unreachable) {
+ if (leaderAddress.equals(m.address())) {
+ log.info("{}: Leader {} is unreachable", logName(), leaderAddress);
+ return false;
+ }
+ }
+
+ for (Member m: state.getMembers()) {
+ if (leaderAddress.equals(m.address())) {
+ if (m.status() == MemberStatus.up() || m.status() == MemberStatus.weaklyUp()) {
+ log.debug("{}: Leader {} cluster status is {} - leader is available", logName(),
+ leaderAddress, m.status());
+ return true;
+ } else {
+ log.debug("{}: Leader {} cluster status is {} - leader is unavailable", logName(),
+ leaderAddress, m.status());
+ return false;
+ }
+ }
+ }
+
+ log.debug("{}: Leader {} not found in the cluster member set", logName(), leaderAddress);
+
+ 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();