YugabyteDB (2.13.1.0-b60, 21121d69985fbf76aa6958d8f04a9bfa936293b5)

// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except

// in compliance with the License.  You may obtain a copy of the License at

//

// http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing, software distributed under the License

// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express

// or implied.  See the License for the specific language governing permissions and limitations

// under the License.

//

#include "yb/consensus/raft_consensus.h"

#include <algorithm>

#include <memory>

#include <mutex>

#include <boost/optional.hpp>

#include <gflags/gflags.h>

#include "yb/common/wire_protocol.h"

#include "yb/consensus/consensus.pb.h"

#include "yb/consensus/consensus_context.h"

#include "yb/consensus/consensus_peers.h"

#include "yb/consensus/consensus_round.h"

#include "yb/consensus/leader_election.h"

#include "yb/consensus/log.h"

#include "yb/consensus/peer_manager.h"

#include "yb/consensus/quorum_util.h"

#include "yb/consensus/replica_state.h"

#include "yb/consensus/state_change_context.h"

#include "yb/gutil/casts.h"

#include "yb/gutil/map-util.h"

#include "yb/gutil/stringprintf.h"

#include "yb/rpc/messenger.h"

#include "yb/rpc/periodic.h"

#include "yb/rpc/rpc_controller.h"

#include "yb/server/clock.h"

#include "yb/util/debug-util.h"

#include "yb/util/debug/long_operation_tracker.h"

#include "yb/util/debug/trace_event.h"

#include "yb/util/enums.h"

#include "yb/util/flag_tags.h"

#include "yb/util/format.h"

#include "yb/util/logging.h"

#include "yb/util/metrics.h"

#include "yb/util/net/dns_resolver.h"

#include "yb/util/random.h"

#include "yb/util/random_util.h"

#include "yb/util/scope_exit.h"

#include "yb/util/status_format.h"

#include "yb/util/status_log.h"

#include "yb/util/threadpool.h"

#include "yb/util/tostring.h"

#include "yb/util/trace.h"

#include "yb/util/tsan_util.h"

#include "yb/util/url-coding.h"

using namespace std::literals;

using namespace std::placeholders;

DEFINE_int32(raft_heartbeat_interval_ms, yb::NonTsanVsTsan(500, 1000),

             "The heartbeat interval for Raft replication. The leader produces heartbeats "

             "to followers at this interval. The followers expect a heartbeat at this interval "

             "and consider a leader to have failed if it misses several in a row.");

TAG_FLAG(raft_heartbeat_interval_ms, advanced);

DEFINE_double(leader_failure_max_missed_heartbeat_periods, 6.0,

              "Maximum heartbeat periods that the leader can fail to heartbeat in before we "

              "consider the leader to be failed. The total failure timeout in milliseconds is "

              "raft_heartbeat_interval_ms times leader_failure_max_missed_heartbeat_periods. "

              "The value passed to this flag may be fractional.");

TAG_FLAG(leader_failure_max_missed_heartbeat_periods, advanced);

DEFINE_int32(leader_failure_exp_backoff_max_delta_ms, 20 * 1000,

             "Maximum time to sleep in between leader election retries, in addition to the "

             "regular timeout. When leader election fails the interval in between retries "

             "increases exponentially, up to this value.");

TAG_FLAG(leader_failure_exp_backoff_max_delta_ms, experimental);

DEFINE_bool(enable_leader_failure_detection, true,

            "Whether to enable failure detection of tablet leaders. If enabled, attempts will be "

            "made to elect a follower as a new leader when the leader is detected to have failed.");

TAG_FLAG(enable_leader_failure_detection, unsafe);

DEFINE_test_flag(bool, do_not_start_election_test_only, false,

                 "Do not start election even if leader failure is detected. ");

TAG_FLAG(TEST_do_not_start_election_test_only, runtime);

DEFINE_bool(evict_failed_followers, true,

            "Whether to evict followers from the Raft config that have fallen "

            "too far behind the leader's log to catch up normally or have been "

            "unreachable by the leader for longer than "

            "follower_unavailable_considered_failed_sec");

TAG_FLAG(evict_failed_followers, advanced);

DEFINE_test_flag(bool, follower_reject_update_consensus_requests, false,

                 "Whether a follower will return an error for all UpdateConsensus() requests.");

DEFINE_test_flag(bool, follower_pause_update_consensus_requests, false,

                 "Whether a follower will pause all UpdateConsensus() requests.");

DEFINE_test_flag(int32, follower_reject_update_consensus_requests_seconds, 0,

                 "Whether a follower will return an error for all UpdateConsensus() requests for "

                 "the first TEST_follower_reject_update_consensus_requests_seconds seconds after "

                 "the Consensus objet is created.");

DEFINE_test_flag(bool, follower_fail_all_prepare, false,

                 "Whether a follower will fail preparing all operations.");

DEFINE_int32(after_stepdown_delay_election_multiplier, 5,

             "After a peer steps down as a leader, the factor with which to multiply "

             "leader_failure_max_missed_heartbeat_periods to get the delay time before starting a "

             "new election.");

TAG_FLAG(after_stepdown_delay_election_multiplier, advanced);

TAG_FLAG(after_stepdown_delay_election_multiplier, hidden);

DECLARE_int32(memory_limit_warn_threshold_percentage);

DEFINE_test_flag(int32, inject_delay_leader_change_role_append_secs, 0,

                 "Amount of time to delay leader from sending replicate of change role.");

DEFINE_test_flag(double, return_error_on_change_config, 0.0,

                 "Fraction of the time when ChangeConfig will return an error.");

METRIC_DEFINE_counter(tablet, follower_memory_pressure_rejections,

                      "Follower Memory Pressure Rejections",

                      yb::MetricUnit::kRequests,

                      "Number of RPC requests rejected due to "

                      "memory pressure while FOLLOWER.");

METRIC_DEFINE_gauge_int64(tablet, raft_term,

                          "Current Raft Consensus Term",

                          yb::MetricUnit::kUnits,

                          "Current Term of the Raft Consensus algorithm. This number increments "

                          "each time a leader election is started.");

METRIC_DEFINE_lag(tablet, follower_lag_ms,

                  "Follower lag from leader",

                  "The amount of time since the last UpdateConsensus request from the "

                  "leader.", {0, yb::AggregationFunction::kMax} /* optional_args */);

METRIC_DEFINE_gauge_int64(tablet, is_raft_leader,

                          "Is tablet raft leader",

                          yb::MetricUnit::kUnits,

                          "Keeps track whether tablet is raft leader"

                          "1 indicates that the tablet is raft leader");

METRIC_DEFINE_coarse_histogram(

  table, dns_resolve_latency_during_update_raft_config,

  "yb.consensus.RaftConsensus.UpdateRaftConfig DNS Resolve",

  yb::MetricUnit::kMicroseconds,

  "Microseconds spent resolving DNS requests during RaftConsensus::UpdateRaftConfig");

DEFINE_int32(leader_lease_duration_ms, yb::consensus::kDefaultLeaderLeaseDurationMs,

             "Leader lease duration. A leader keeps establishing a new lease or extending the "

             "existing one with every UpdateConsensus. A new server is not allowed to serve as a "

             "leader (i.e. serve up-to-date read requests or acknowledge write requests) until a "

             "lease of this duration has definitely expired on the old leader's side.");

DEFINE_int32(ht_lease_duration_ms, 2000,

             "Hybrid time leader lease duration. A leader keeps establishing a new lease or "

             "extending the existing one with every UpdateConsensus. A new server is not allowed "

             "to add entries to RAFT log until a lease of the old leader is expired. 0 to disable."

             );

DEFINE_int32(min_leader_stepdown_retry_interval_ms,

             20 * 1000,

             "Minimum amount of time between successive attempts to perform the leader stepdown "

             "for the same combination of tablet and intended (target) leader. This is needed "

             "to avoid infinite leader stepdown loops when the current leader never has a chance "

             "to update the intended leader with its latest records.");

DEFINE_bool(use_preelection, true, "Whether to use pre election, before doing actual election.");

DEFINE_int32(temporary_disable_preelections_timeout_ms, 10 * 60 * 1000,

             "If some of nodes does not support preelections, then we disable them for this "

             "amount of time.");

DEFINE_test_flag(bool, pause_update_replica, false,

                 "Pause RaftConsensus::UpdateReplica processing before snoozing failure detector.");

DEFINE_test_flag(bool, pause_update_majority_replicated, false,

                 "Pause RaftConsensus::UpdateMajorityReplicated.");

DEFINE_test_flag(int32, log_change_config_every_n, 1,

                 "How often to log change config information. "

                 "Used to reduce the number of lines being printed for change config requests "

                 "when a test simulates a failure that would generate a log of these requests.");

DEFINE_bool(enable_lease_revocation, true, "Enables lease revocation mechanism");

DEFINE_bool(quick_leader_election_on_create, false,

            "Do we trigger quick leader elections on table creation.");

TAG_FLAG(quick_leader_election_on_create, advanced);

TAG_FLAG(quick_leader_election_on_create, hidden);

DEFINE_bool(

    stepdown_disable_graceful_transition, false,

    "During a leader stepdown, disable graceful leadership transfer "

    "to an up to date peer");

DEFINE_bool(

    raft_disallow_concurrent_outstanding_report_failure_tasks, true,

    "If true, only submit a new report failure task if there is not one outstanding.");

TAG_FLAG(raft_disallow_concurrent_outstanding_report_failure_tasks, advanced);

TAG_FLAG(raft_disallow_concurrent_outstanding_report_failure_tasks, hidden);

DEFINE_int64(protege_synchronization_timeout_ms, 1000,

             "Timeout to synchronize protege before performing step down. "

             "0 to disable synchronization.");

namespace yb {

namespace consensus {

using log::LogEntryBatch;

using rpc::PeriodicTimer;

using std::shared_ptr;

using std::unique_ptr;

using std::weak_ptr;

using strings::Substitute;

using tserver::TabletServerErrorPB;

namespace {

const RaftPeerPB* FindPeer(const RaftConfigPB& active_config, const std::string& uuid) {

  for (const RaftPeerPB& peer : active_config.peers()) {

    if (peer.permanent_uuid() == uuid) {

      return &peer;

    }

  }

  return nullptr;

}

// Helper function to check if the op is a non-Operation op.

bool IsConsensusOnlyOperation(OperationType op_type) {

  return op_type == NO_OP || 
op_type == CHANGE_CONFIG_OP14.2M
;

}

// Helper to check if the op is Change Config op.

bool IsChangeConfigOperation(OperationType op_type) {

  return op_type == CHANGE_CONFIG_OP;

}

class NonTrackedRoundCallback : public ConsensusRoundCallback {

 public:

  explicit NonTrackedRoundCallback(ConsensusRound* round, const StdStatusCallback& callback)

      : round_(round), callback_(callback) {

  }

  void AddedToLeader(const OpId& op_id, const OpId& committed_op_id) override {

    auto& replicate_msg = *round_->replicate_msg();

    op_id.ToPB(replicate_msg.mutable_id());

    committed_op_id.ToPB(replicate_msg.mutable_committed_op_id());

  }

  void ReplicationFinished(

      const Status& status, int64_t leader_term, OpIds* applied_op_ids) override {

    down_cast<RaftConsensus*>(round_->consensus())->NonTrackedRoundReplicationFinished(

        round_, callback_, status);

  }

 private:

  ConsensusRound* round_;

  StdStatusCallback callback_;

};

} // namespace

std::unique_ptr<ConsensusRoundCallback> MakeNonTrackedRoundCallback(

    ConsensusRound* round, const StdStatusCallback& callback) {

  return std::make_unique<NonTrackedRoundCallback>(round, callback);

}

struct RaftConsensus::LeaderRequest {

  std::string leader_uuid;

  yb::OpId preceding_op_id;

  yb::OpId committed_op_id;

  ReplicateMsgs messages;

  // The positional index of the first message selected to be appended, in the

  // original leader's request message sequence.

  int64_t first_message_idx;

  std::string OpsRangeString() const;

};

shared_ptr<RaftConsensus> RaftConsensus::Create(

    const ConsensusOptions& options,

    std::unique_ptr<ConsensusMetadata> cmeta,

    const RaftPeerPB& local_peer_pb,

    const scoped_refptr<MetricEntity>& table_metric_entity,

    const scoped_refptr<MetricEntity>& tablet_metric_entity,

    const scoped_refptr<server::Clock>& clock,

    ConsensusContext* consensus_context,

    rpc::Messenger* messenger,

    rpc::ProxyCache* proxy_cache,

    const scoped_refptr<log::Log>& log,

    const shared_ptr<MemTracker>& server_mem_tracker,

    const shared_ptr<MemTracker>& parent_mem_tracker,

    const Callback<void(std::shared_ptr<StateChangeContext> context)> mark_dirty_clbk,

    TableType table_type,

    ThreadPool* raft_pool,

    RetryableRequests* retryable_requests,

    MultiRaftManager* multi_raft_manager) {

  auto rpc_factory = std::make_unique<RpcPeerProxyFactory>(

    messenger, proxy_cache, local_peer_pb.cloud_info());

  // The message queue that keeps track of which operations need to be replicated

  // where.

  auto queue = std::make_unique<PeerMessageQueue>(

      tablet_metric_entity,

      log,

      server_mem_tracker,

      parent_mem_tracker,

      local_peer_pb,

      options.tablet_id,

      clock,

      consensus_context,

      raft_pool->NewToken(ThreadPool::ExecutionMode::SERIAL));

  DCHECK(local_peer_pb.has_permanent_uuid());

  const string& peer_uuid = local_peer_pb.permanent_uuid();

  // A single Raft thread pool token is shared between RaftConsensus and

  // PeerManager. Because PeerManager is owned by RaftConsensus, it receives a

  // raw pointer to the token, to emphasize that RaftConsensus is responsible

  // for destroying the token.

  unique_ptr<ThreadPoolToken> raft_pool_token(raft_pool->NewToken(

      ThreadPool::ExecutionMode::CONCURRENT));

  // A manager for the set of peers that actually send the operations both remotely

  // and to the local wal.

  auto peer_manager = std::make_unique<PeerManager>(

      options.tablet_id,

      peer_uuid,

      rpc_factory.get(),

      queue.get(),

      raft_pool_token.get(),

      multi_raft_manager);

  return std::make_shared<RaftConsensus>(

      options,

      std::move(cmeta),

      std::move(rpc_factory),

      std::move(queue),

      std::move(peer_manager),

      std::move(raft_pool_token),

      table_metric_entity,

      tablet_metric_entity,

      peer_uuid,

      clock,

      consensus_context,

      log,

      parent_mem_tracker,

      mark_dirty_clbk,

      table_type,

      retryable_requests);

}

RaftConsensus::RaftConsensus(

    const ConsensusOptions& options, std::unique_ptr<ConsensusMetadata> cmeta,

    std::unique_ptr<PeerProxyFactory> proxy_factory,

    std::unique_ptr<PeerMessageQueue> queue,

    std::unique_ptr<PeerManager> peer_manager,

    std::unique_ptr<ThreadPoolToken> raft_pool_token,

    const scoped_refptr<MetricEntity>& table_metric_entity,

    const scoped_refptr<MetricEntity>& tablet_metric_entity,

    const std::string& peer_uuid, const scoped_refptr<server::Clock>& clock,

    ConsensusContext* consensus_context, const scoped_refptr<log::Log>& log,

    shared_ptr<MemTracker> parent_mem_tracker,

    Callback<void(std::shared_ptr<StateChangeContext> context)> mark_dirty_clbk,

    TableType table_type,

    RetryableRequests* retryable_requests)

    : raft_pool_token_(std::move(raft_pool_token)),

      log_(log),

      clock_(clock),

      peer_proxy_factory_(std::move(proxy_factory)),

      peer_manager_(std::move(peer_manager)),

      queue_(std::move(queue)),

      rng_(GetRandomSeed32()),

      withhold_votes_until_(MonoTime::Min()),

      step_down_check_tracker_(&peer_proxy_factory_->messenger()->scheduler()),

      mark_dirty_clbk_(std::move(mark_dirty_clbk)),

      shutdown_(false),

      follower_memory_pressure_rejections_(tablet_metric_entity->FindOrCreateCounter(

          &METRIC_follower_memory_pressure_rejections)),

      term_metric_(tablet_metric_entity->FindOrCreateGauge(&METRIC_raft_term,

                                                    cmeta->current_term())),

      follower_last_update_time_ms_metric_(

          tablet_metric_entity->FindOrCreateAtomicMillisLag(&METRIC_follower_lag_ms)),

      is_raft_leader_metric_(tablet_metric_entity->FindOrCreateGauge(&METRIC_is_raft_leader,

                                                              static_cast<int64_t>(0))),

      parent_mem_tracker_(std::move(parent_mem_tracker)),

      table_type_(table_type),

      update_raft_config_dns_latency_(

          METRIC_dns_resolve_latency_during_update_raft_config.Instantiate(table_metric_entity)),

      split_parent_tablet_id_(

          cmeta->has_split_parent_tablet_id() ? cmeta->split_parent_tablet_id() : "") {

  DCHECK_NOTNULL(log_.get());

  if (PREDICT_FALSE(FLAGS_TEST_follower_reject_update_consensus_requests_seconds > 0)) {

    withold_replica_updates_until_ = MonoTime::Now() +

        MonoDelta::FromSeconds(FLAGS_TEST_follower_reject_update_consensus_requests_seconds);

  }

  state_ = std::make_unique<ReplicaState>(

      options,

      peer_uuid,

      std::move(cmeta),

      DCHECK_NOTNULL(consensus_context),

      this,

      retryable_requests,

      std::bind(&PeerMessageQueue::TrackOperationsMemory, queue_.get(), _1));

  peer_manager_->SetConsensus(this);

}

RaftConsensus::~RaftConsensus() {

  Shutdown();

}

Status RaftConsensus::Start(const ConsensusBootstrapInfo& info) {

  RETURN_NOT_OK(ExecuteHook(PRE_START));

  // Capture a weak_ptr reference into the functor so it can safely handle

  // outliving the consensus instance.

  std::weak_ptr<RaftConsensus> w = shared_from_this();

  failure_detector_ = PeriodicTimer::Create(

      peer_proxy_factory_->messenger(),

      [w]() {

        if (auto consensus = w.lock()) {

          consensus->ReportFailureDetected();

        }

      },

      MinimumElectionTimeout());

  {

    ReplicaState::UniqueLock lock;

    RETURN_NOT_OK(state_->LockForStart(&lock));

    state_->ClearLeaderUnlocked();

    RETURN_NOT_OK_PREPEND(state_->StartUnlocked(info.last_id),

                          "Unable to start RAFT ReplicaState");

    LOG_WITH_PREFIX(INFO) << "Replica starting. Triggering "

                          << info.orphaned_replicates.size()

                          << " pending operations. Active config: "

                          << state_->GetActiveConfigUnlocked().ShortDebugString();

    for (const auto& replicate : info.orphaned_replicates) {

      ReplicateMsgPtr replicate_ptr = std::make_shared<ReplicateMsg>(*replicate);

      RETURN_NOT_OK(StartReplicaOperationUnlocked(replicate_ptr, HybridTime::kInvalid));

    }

    RETURN_NOT_OK(state_->InitCommittedOpIdUnlocked(yb::OpId::FromPB(info.last_committed_id)));

    queue_->Init(state_->GetLastReceivedOpIdUnlocked());

  }

  {

    ReplicaState::UniqueLock lock;

    RETURN_NOT_OK(state_->LockForConfigChange(&lock));

    // If this is the first term expire the FD immediately so that we have a fast first

    // election, otherwise we just let the timer expire normally.

    MonoDelta initial_delta = MonoDelta();

    if (state_->GetCurrentTermUnlocked() == 0) {

      // The failure detector is initialized to a low value to trigger an early election

      // (unless someone else requested a vote from us first, which resets the

      // election timer). We do it this way instead of immediately running an

      // election to get a higher likelihood of enough servers being available

      // when the first one attempts an election to avoid multiple election

      // cycles on startup, while keeping that "waiting period" random. If there is only one peer,

      // trigger an election right away.

      if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {

        LOG_WITH_PREFIX(INFO) << "Consensus starting up: Expiring fail detector timer "

                                 "to make a prompt election more likely";

        // Gating quick leader elections on table creation since prompter leader elections are

        // more likely to fail due to uninitialized peers or conflicting elections, which could

        // have unforseen consequences.

        if (FLAGS_quick_leader_election_on_create) {

          initial_delta = (state_->GetCommittedConfigUnlocked().peers_size() == 1) ?

              MonoDelta::kZero :

              MonoDelta::FromMilliseconds(rng_.Uniform(FLAGS_raft_heartbeat_interval_ms));

        }

      }

    }

    RETURN_NOT_OK(BecomeReplicaUnlocked(std::string(), initial_delta));

  }

  RETURN_NOT_OK(ExecuteHook(POST_START));

  // The context tracks that the current caller does not hold the lock for consensus state.

  // So mark dirty callback, e.g., consensus->ConsensusState() for master consensus callback of

  // SysCatalogStateChanged, can get the lock when needed.

  auto context = std::make_shared<StateChangeContext>(StateChangeReason::CONSENSUS_STARTED, false);

  // Report become visible to the Master.

  MarkDirty(context);

  return Status::OK();

}

bool RaftConsensus::IsRunning() const {

  auto lock = state_->LockForRead();

  return state_->state() == ReplicaState::kRunning;

}

Status RaftConsensus::EmulateElection() {

  ReplicaState::UniqueLock lock;

  RETURN_NOT_OK(state_->LockForConfigChange(&lock));

  LOG_WITH_PREFIX(INFO) << "Emulating election...";

  // Assume leadership of new term.

  RETURN_NOT_OK(IncrementTermUnlocked());

  SetLeaderUuidUnlocked(state_->GetPeerUuid());

  return BecomeLeaderUnlocked();

}

Status RaftConsensus::DoStartElection(const LeaderElectionData& data, PreElected preelected) {

  TRACE_EVENT2("consensus", "RaftConsensus::StartElection",

               "peer", peer_uuid(),

               "tablet", tablet_id());

  
VLOG_WITH_PREFIX_AND_FUNC47
(1) << data.ToString()47
;

  if (FLAGS_TEST_do_not_start_election_test_only) {

    LOG(INFO) << "Election start skipped as TEST_do_not_start_election_test_only flag "

                 "is set to true.";

    return Status::OK();

  }

  // If pre-elections disabled or we already won pre-election then start regular election,

  // otherwise pre-election is started.

  // Pre-elections could be disable via flag, or temporarily if some nodes do not support them.

  auto preelection = ANNOTATE_UNPROTECTED_READ(FLAGS_use_preelection) && 
!preelected1.22M
 &&

                     
disable_pre_elections_until_ < CoarseMonoClock::now()674k
;

  const char* election_name = preelection ? 
"pre-election"675k
 : 
"election"550k
;

  LeaderElectionPtr election;

  {

    ReplicaState::UniqueLock lock;

    RETURN_NOT_OK(state_->LockForConfigChange(&lock));

    if (data.initial_election && 
state_->GetCurrentTermUnlocked() != 089.5k
) {

      LOG_WITH_PREFIX(INFO) << "Not starting initial " << election_name << " -- non zero term";

      return Status::OK();

    }

    PeerRole active_role = state_->GetActiveRoleUnlocked();

    if (active_role == PeerRole::LEADER) {

      LOG_WITH_PREFIX(INFO) << "Not starting " << election_name << " -- already leader";

      return Status::OK();

    }

    
if (1.22M
active_role == PeerRole::LEARNER1.22M
 || active_role == PeerRole::READ_REPLICA) {

      LOG_WITH_PREFIX(INFO) << "Not starting " << election_name << " -- role is " << active_role

                            << ", pending = " << state_->IsConfigChangePendingUnlocked()

                            << ", active_role=" << active_role;

      return Status::OK();

    }

    if (PREDICT_FALSE(active_role == PeerRole::NON_PARTICIPANT)) {

      
VLOG_WITH_PREFIX0
(1) << "Not starting " << election_name << " -- non participant"0
;

      // Avoid excessive election noise while in this state.

      SnoozeFailureDetector(DO_NOT_LOG);

      return STATUS_FORMAT(

          IllegalState,

          "Not starting $0: Node is currently a non-participant in the raft config: $1",

          election_name, state_->GetActiveConfigUnlocked());

    }

    // Default is to start the election now. But if we are starting a pending election, see if

    // there is an op id pending upon indeed and if it has been committed to the log. The op id

    // could have been cleared if the pending election has already been started or another peer

    // has jumped before we can start.

    bool start_now = true;

    if (data.pending_commit) {

      const auto required_id = !data.must_be_committed_opid

          ? state_->GetPendingElectionOpIdUnlocked() : data.must_be_committed_opid;

      const Status advance_committed_index_status = ResultToStatus(

          state_->AdvanceCommittedOpIdUnlocked(required_id, CouldStop::kFalse));

      if (!advance_committed_index_status.ok()) {

        LOG(WARNING) << "Starting an " << election_name << " but the latest committed OpId is not "

                        "present in this peer's log: "

                     << required_id << ". " << "Status: " << advance_committed_index_status;

      }

      start_now = required_id.index <= state_->GetCommittedOpIdUnlocked().index;

    }

    if (
start_now1.22M
) {

      if (state_->HasLeaderUnlocked()) {

        LOG_WITH_PREFIX(INFO) << "Fail of leader " << state_->GetLeaderUuidUnlocked()

                              << " detected. Triggering leader " << election_name

                              << ", mode=" << data.mode;

      } else {

        LOG_WITH_PREFIX(INFO) << "Triggering leader " << election_name << ", mode=" << data.mode;

      }

      // Snooze to avoid the election timer firing again as much as possible.

      // We do not disable the election timer while running an election.

      MonoDelta timeout = LeaderElectionExpBackoffDeltaUnlocked();

      SnoozeFailureDetector(ALLOW_LOGGING, timeout);

      election = 
VERIFY_RESULT737k
(CreateElectionUnlocked(data, timeout, PreElection(preelection)));

    } else if (data.pending_commit && 
!data.must_be_committed_opid.empty()0
) {

      // Queue up the pending op id if specified.

      state_->SetPendingElectionOpIdUnlocked(data.must_be_committed_opid);

      LOG_WITH_PREFIX(INFO)

          << "Leader " << election_name << " is pending upon log commitment of OpId "

          << data.must_be_committed_opid;

    } else {

      LOG_WITH_PREFIX(INFO) << "Ignore " << __func__ << " existing wait on op id";

    }

  }

  // Start the election outside the lock.

  
if (737k
election737k
) {

    election->Run();

  }

  return Status::OK();

}

Result<LeaderElectionPtr> RaftConsensus::CreateElectionUnlocked(

    const LeaderElectionData& data, MonoDelta timeout, PreElection preelection) {

  int64_t new_term;

  if (preelection) {

    new_term = state_->GetCurrentTermUnlocked() + 1;

  } else {

    // Increment the term.

    RETURN_NOT_OK(IncrementTermUnlocked());

    new_term = state_->GetCurrentTermUnlocked();

  }

  const RaftConfigPB& active_config = state_->GetActiveConfigUnlocked();

  LOG_WITH_PREFIX(INFO) << "Starting " << (preelection ? 
"pre-"674k
 : 
""62.8k
) << "election with config: "

                        << active_config.ShortDebugString();

  // Initialize the VoteCounter.

  auto num_voters = CountVoters(active_config);

  auto majority_size = MajoritySize(num_voters);

  // Vote for ourselves.

  if (!preelection) {

    // TODO: Consider using a separate Mutex for voting, which must sync to disk.

    RETURN_NOT_OK(state_->SetVotedForCurrentTermUnlocked(state_->GetPeerUuid()));

  }

  auto counter = std::make_unique<VoteCounter>(num_voters, majority_size);

  bool duplicate;

  RETURN_NOT_OK(counter->RegisterVote(state_->GetPeerUuid(), ElectionVote::kGranted, &duplicate));

  CHECK(!duplicate) << state_->LogPrefix()

                    << "Inexplicable duplicate self-vote for term "

                    << state_->GetCurrentTermUnlocked();

  VoteRequestPB request;

  request.set_ignore_live_leader(data.mode == ElectionMode::ELECT_EVEN_IF_LEADER_IS_ALIVE);

  request.set_candidate_uuid(state_->GetPeerUuid());

  request.set_candidate_term(new_term);

  request.set_tablet_id(state_->GetOptions().tablet_id);

  request.set_preelection(preelection);

  state_->GetLastReceivedOpIdUnlocked().ToPB(

      request.mutable_candidate_status()->mutable_last_received());

  LeaderElectionPtr result(new LeaderElection(

      active_config,

      peer_proxy_factory_.get(),

      request,

      std::move(counter),

      timeout,

      preelection,

      data.suppress_vote_request,

      std::bind(&RaftConsensus::ElectionCallback, shared_from_this(), data, _1)));

  if (!preelection) {

    // Clear the pending election op id so that we won't start the same pending election again.

    // Pre-election does not change state, so should not do it in this case.

    state_->ClearPendingElectionOpIdUnlocked();

  }

  return result;

}

Status RaftConsensus::WaitUntilLeaderForTests(const MonoDelta& timeout) {

  MonoTime deadline = MonoTime::Now();

  deadline.AddDelta(timeout);

  while (MonoTime::Now().ComesBefore(deadline)) {

    if (GetLeaderStatus() == LeaderStatus::LEADER_AND_READY) {

      return Status::OK();

    }

    SleepFor(MonoDelta::FromMilliseconds(10));

  }

  return STATUS(TimedOut, Substitute("Peer $0 is not leader of tablet $1 after $2. Role: $3",

                                     peer_uuid(), tablet_id(), timeout.ToString(), role()));

}

string RaftConsensus::ServersInTransitionMessage() {

  string err_msg;

  const RaftConfigPB& active_config = state_->GetActiveConfigUnlocked();

  const RaftConfigPB& committed_config = state_->GetCommittedConfigUnlocked();

  auto servers_in_transition = CountServersInTransition(active_config);

  auto committed_servers_in_transition = CountServersInTransition(committed_config);

  LOG(INFO) << Substitute("Active config has $0 and committed has $1 servers in transition.",

                          servers_in_transition, committed_servers_in_transition);

  if (servers_in_transition != 0 || 
committed_servers_in_transition != 011.1k
) {

    err_msg = Substitute("Leader not ready to step down as there are $0 active config peers"

                         " in transition, $1 in committed. Configs:\nactive=$2\ncommit=$3",

                         servers_in_transition, committed_servers_in_transition,

                         active_config.ShortDebugString(), committed_config.ShortDebugString());

    LOG(INFO) << err_msg;

  }

  return err_msg;

}

Status RaftConsensus::StartStepDownUnlocked(const RaftPeerPB& peer, bool graceful) {

  auto election_state = std::make_shared<RunLeaderElectionState>();

  election_state->proxy = peer_proxy_factory_->NewProxy(peer);

  election_state->req.set_originator_uuid(state_->GetPeerUuid());

  election_state->req.set_dest_uuid(peer.permanent_uuid());

  election_state->req.set_tablet_id(state_->GetOptions().tablet_id);

  election_state->rpc.set_invoke_callback_mode(rpc::InvokeCallbackMode::kThreadPoolHigh);

  election_state->proxy->RunLeaderElectionAsync(

      &election_state->req, &election_state->resp, &election_state->rpc,

      std::bind(&RaftConsensus::RunLeaderElectionResponseRpcCallback, this,

          election_state));

  LOG_WITH_PREFIX(INFO) << "Transferring leadership to " << peer.permanent_uuid();

  return BecomeReplicaUnlocked(

      graceful ? 
std::string()569
 : 
peer.permanent_uuid()9.43k
, MonoDelta());

}

void RaftConsensus::CheckDelayedStepDown(const Status& status) {

  ReplicaState::UniqueLock lock;

  auto lock_status = state_->LockForConfigChange(&lock);

  if (!lock_status.ok()) {

    LOG_WITH_PREFIX(INFO) << "Failed to check delayed election: " << lock_status;

    return;

  }

  if (state_->GetCurrentTermUnlocked() != delayed_step_down_.term) {

    return;

  }

  const auto& config = state_->GetActiveConfigUnlocked();

  const auto* peer = FindPeer(config, delayed_step_down_.protege);

  if (peer) {

    LOG_WITH_PREFIX(INFO) << "Step down in favor on not synchronized protege: "

                          << delayed_step_down_.protege;

    WARN_NOT_OK(StartStepDownUnlocked(*peer, delayed_step_down_.graceful),

                "Start step down failed");

  } else {

    LOG_WITH_PREFIX(INFO) << "Failed to synchronize with protege " << delayed_step_down_.protege

                          << " and cannot find it in config: " << config.ShortDebugString();

    delayed_step_down_.term = OpId::kUnknownTerm;

  }

}

Status RaftConsensus::StepDown(const LeaderStepDownRequestPB* req, LeaderStepDownResponsePB* resp) {

  TRACE_EVENT0("consensus", "RaftConsensus::StepDown");

  ReplicaState::UniqueLock lock;

  RETURN_NOT_OK(state_->LockForConfigChange(&lock));

  // A sanity check that this request was routed to the correct RaftConsensus.

  const auto& tablet_id = req->tablet_id();

  if (tablet_id != this->tablet_id()) {

    resp->mutable_error()->set_code(TabletServerErrorPB::UNKNOWN_ERROR);

    const auto msg = Format(

        "Received a leader stepdown operation for wrong tablet id: $0, must be: $1",

        tablet_id, this->tablet_id());

    LOG_WITH_PREFIX(ERROR) << msg;

    StatusToPB(STATUS(IllegalState, msg), resp->mutable_error()->mutable_status());

    return Status::OK();

  }

  if (state_->GetActiveRoleUnlocked() != PeerRole::LEADER) {

    resp->mutable_error()->set_code(TabletServerErrorPB::NOT_THE_LEADER);

    StatusToPB(STATUS(IllegalState, "Not currently leader"),

               resp->mutable_error()->mutable_status());

    // We return OK so that the tablet service won't overwrite the error code.

    return Status::OK();

  }

  // The leader needs to be ready to perform a step down. There should be no PRE_VOTER in both

  // active and committed configs - ENG-557.

  const string err_msg = ServersInTransitionMessage();

  if (!err_msg.empty()) {

    resp->mutable_error()->set_code(TabletServerErrorPB::LEADER_NOT_READY_TO_STEP_DOWN);

    StatusToPB(STATUS(IllegalState, err_msg), resp->mutable_error()->mutable_status());

    return Status::OK();

  }

  std::string new_leader_uuid;

  // If a new leader is nominated, find it among peers to send RunLeaderElection request.

  // See https://ramcloud.stanford.edu/~ongaro/thesis.pdf, section 3.10 for this mechanism

  // to transfer the leadership.

  const bool forced = (req->has_force_step_down() && 
req->force_step_down()0
);

  if (req->has_new_leader_uuid()) {

    new_leader_uuid = req->new_leader_uuid();

    if (!forced && !queue_->CanPeerBecomeLeader(new_leader_uuid)) {

      resp->mutable_error()->set_code(TabletServerErrorPB::LEADER_NOT_READY_TO_STEP_DOWN);

      StatusToPB(

          STATUS(IllegalState, "Suggested peer is not caught up yet"),

          resp->mutable_error()->mutable_status());

      // We return OK so that the tablet service won't overwrite the error code.

      return Status::OK();

    }

  }

  bool graceful_stepdown = false;

  if (new_leader_uuid.empty() && 
!FLAGS_stepdown_disable_graceful_transition589
 &&

      
!(589
req->has_disable_graceful_transition()589
 && 
req->disable_graceful_transition()1
)) {

    new_leader_uuid = queue_->GetUpToDatePeer();

    LOG_WITH_PREFIX(INFO) << "Selected up to date candidate protege leader [" << new_leader_uuid

                          << "]";

    graceful_stepdown = true;

  }

  const auto& local_peer_uuid = state_->GetPeerUuid();

  if (
!new_leader_uuid.empty()10.9k
) {

    const auto leadership_transfer_description =

        Format("tablet $0 from $1 to $2", tablet_id, local_peer_uuid, new_leader_uuid);

    if (
!forced10.9k
 && new_leader_uuid == protege_leader_uuid_ && 
election_lost_by_protege_at_1.08k
) {

      const MonoDelta time_since_election_loss_by_protege =

          MonoTime::Now() - election_lost_by_protege_at_;

      if (time_since_election_loss_by_protege.ToMilliseconds() <

              FLAGS_min_leader_stepdown_retry_interval_ms) {

        LOG_WITH_PREFIX(INFO) << "Unable to execute leadership transfer for "

                              << leadership_transfer_description

                              << " because the intended leader already lost an election only "

                              << ToString(time_since_election_loss_by_protege) << " ago (within "

                              << FLAGS_min_leader_stepdown_retry_interval_ms << " ms).";

        if (req->has_new_leader_uuid()) {

          LOG_WITH_PREFIX(INFO) << "Rejecting leader stepdown request for "

                                << leadership_transfer_description;

          resp->mutable_error()->set_code(TabletServerErrorPB::LEADER_NOT_READY_TO_STEP_DOWN);

          resp->set_time_since_election_failure_ms(

              time_since_election_loss_by_protege.ToMilliseconds());

          StatusToPB(

              STATUS(IllegalState, "Suggested peer lost an election recently"),

              resp->mutable_error()->mutable_status());

          // We return OK so that the tablet service won't overwrite the error code.

          return Status::OK();

        } else {

          // we were attempting a graceful transfer of our own choice

          // which is no longer possible

          new_leader_uuid.clear();

        }

      }

      election_lost_by_protege_at_ = MonoTime();

    }

  }

  if (!new_leader_uuid.empty()) {

    const auto* peer = FindPeer(state_->GetActiveConfigUnlocked(), new_leader_uuid);

    if (
peer10.0k
 && peer->member_type() == PeerMemberType::VOTER) {

      auto timeout_ms = FLAGS_protege_synchronization_timeout_ms;

      if (timeout_ms != 0 &&

          
queue_->PeerLastReceivedOpId(new_leader_uuid) < GetLatestOpIdFromLog()10.0k
) {

        delayed_step_down_ = DelayedStepDown {

          .term = state_->GetCurrentTermUnlocked(),

          .protege = new_leader_uuid,

          .graceful = graceful_stepdown,

        };

        LOG_WITH_PREFIX(INFO) << "Delay step down: " << delayed_step_down_.ToString();

        step_down_check_tracker_.Schedule(

            std::bind(&RaftConsensus::CheckDelayedStepDown, this, _1),

            1ms * timeout_ms);

        return Status::OK();

      }

      return StartStepDownUnlocked(*peer, graceful_stepdown);

    }

    LOG_WITH_PREFIX(WARNING) << "New leader " << new_leader_uuid << " not found.";

    if (req->has_new_leader_uuid()) {

      resp->mutable_error()->set_code(TabletServerErrorPB::LEADER_NOT_READY_TO_STEP_DOWN);

      StatusToPB(

          STATUS(IllegalState, "New leader not found among peers"),

          resp->mutable_error()->mutable_status());

      // We return OK so that the tablet service won't overwrite the error code.

      return Status::OK();

    } else {

      // we were attempting a graceful transfer of our own choice

      // which is no longer possible

      new_leader_uuid.clear();

    }

  }

  
if (15
graceful_stepdown15
) {

    new_leader_uuid.clear();

  }

  RETURN_NOT_OK(BecomeReplicaUnlocked(new_leader_uuid, MonoDelta()));

  return Status::OK();

}

Status RaftConsensus::ElectionLostByProtege(const std::string& election_lost_by_uuid) {

  if (election_lost_by_uuid.empty()) {

    return STATUS(InvalidArgument, "election_lost_by_uuid could not be empty");

  }

  auto start_election = false;

  {

    ReplicaState::UniqueLock lock;

    RETURN_NOT_OK(state_->LockForConfigChange(&lock));

    if (election_lost_by_uuid == protege_leader_uuid_) {

      LOG_WITH_PREFIX(INFO) << "Our protege " << election_lost_by_uuid

                            << ", lost election. Has leader: "

                            << state_->HasLeaderUnlocked();

      withhold_election_start_until_.store(MonoTime::Min(), std::memory_order_relaxed);

      election_lost_by_protege_at_ = MonoTime::Now();

      start_election = !state_->HasLeaderUnlocked();

    }

  }

  if (start_election) {

    return StartElection({ElectionMode::NORMAL_ELECTION});

  }

  return Status::OK();

}

void RaftConsensus::WithholdElectionAfterStepDown(const std::string& protege_uuid) {

  DCHECK(state_->IsLocked());

  protege_leader_uuid_ = protege_uuid;

  auto timeout = MonoDelta::FromMilliseconds(

      FLAGS_leader_failure_max_missed_heartbeat_periods *

      FLAGS_raft_heartbeat_interval_ms);

  if (!protege_uuid.empty()) {

    // Actually we have 2 kinds of step downs.

    // 1) We step down in favor of some protege.

    // 2) We step down because term was advanced or just started.

    // In second case we should not withhold election for a long period of time.

    timeout *= FLAGS_after_stepdown_delay_election_multiplier;

  }

  auto deadline = MonoTime::Now() + timeout;

  VLOG
(2) << "Withholding election for " << timeout0
;

  withhold_election_start_until_.store(deadline, std::memory_order_release);

  election_lost_by_protege_at_ = MonoTime();

}

void RaftConsensus::RunLeaderElectionResponseRpcCallback(

    shared_ptr<RunLeaderElectionState> election_state) {

  // Check for RPC errors.

  if (!election_state->rpc.status().ok()) {

    LOG_WITH_PREFIX(WARNING) << "RPC error from RunLeaderElection() call to peer "

                             << election_state->req.dest_uuid() << ": "

                             << election_state->rpc.status();

  // Check for tablet errors.

  } else if (election_state->resp.has_error()) {

    LOG_WITH_PREFIX(WARNING) << "Tablet error from RunLeaderElection() call to peer "

                             << election_state->req.dest_uuid() << ": "

                             << StatusFromPB(election_state->resp.error().status());

  }

}

void RaftConsensus::ReportFailureDetectedTask() {

  auto scope_exit = ScopeExit([this] {

    outstanding_report_failure_task_.clear(std::memory_order_release);

  });

  MonoTime now;

  for (;;) {

    // Do not start election for an extended period of time if we were recently stepped down.

    auto old_value = withhold_election_start_until_.load(std::memory_order_acquire);

    if (old_value == MonoTime::Min()) {

      break;

    }

    
if (1.39k
!now.Initialized()1.39k
) {

      now = MonoTime::Now();

    }

    if (now < old_value) {

      VLOG
(1) << "Skipping election due to delayed timeout for " << (old_value - now)0
;

      return;

    }

    // If we ever stepped down and then delayed election start did get scheduled, reset that we

    // are out of that extra delay mode.

    if (withhold_election_start_until_.compare_exchange_weak(

        old_value, MonoTime::Min(), std::memory_order_release)) {

      break;

    }

  }

  // Start an election.

  LOG_WITH_PREFIX(INFO) << "ReportFailDetected: Starting NORMAL_ELECTION...";

  Status s = StartElection({ElectionMode::NORMAL_ELECTION});

  if (PREDICT_FALSE(!s.ok())) {

    LOG_WITH_PREFIX(WARNING) << "Failed to trigger leader election: " << s.ToString();

  }

}

void RaftConsensus::ReportFailureDetected() {

  if (FLAGS_raft_disallow_concurrent_outstanding_report_failure_tasks &&

      
outstanding_report_failure_task_.test_and_set(std::memory_order_acq_rel)818k
) {

    VLOG(4)

        << "Returning from ReportFailureDetected as there is already an outstanding report task.";

  } else {

    // We're running on a timer thread; start an election on a different thread pool.

    auto s = raft_pool_token_->SubmitFunc(

        std::bind(&RaftConsensus::ReportFailureDetectedTask, shared_from_this()));

    WARN_NOT_OK(s, "Failed to submit failure detected task");

    if (!s.ok()) {

      outstanding_report_failure_task_.clear(std::memory_order_release);

    }

  }

}

Status RaftConsensus::BecomeLeaderUnlocked() {