text: // 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.
  
  #include "kudu/consensus/raft_consensus.h"
  
  #include <algorithm>
  #include <cmath>
  #include <cstdint>
  #include <functional>
  #include <iterator>
  #include <memory>
  #include <mutex>
  #include <ostream>
  #include <unordered_map>
  #include <unordered_set>
  
  #include <boost/optional/optional.hpp>
  #include <gflags/gflags.h>
  #include <google/protobuf/util/message_differencer.h>
  
  #include "kudu/common/timestamp.h"
  #include "kudu/common/wire_protocol.h"
  #include "kudu/consensus/consensus.pb.h"
  #include "kudu/consensus/consensus_meta.h"
  #include "kudu/consensus/consensus_meta_manager.h"
  #include "kudu/consensus/consensus_peers.h"
  #include "kudu/consensus/leader_election.h"
  #include "kudu/consensus/log.h"
  #include "kudu/consensus/metadata.pb.h"
  #include "kudu/consensus/opid_util.h"
  #include "kudu/consensus/peer_manager.h"
  #include "kudu/consensus/pending_rounds.h"
  #include "kudu/consensus/quorum_util.h"
  #include "kudu/consensus/time_manager.h"
  #include "kudu/gutil/macros.h"
  #include "kudu/gutil/map-util.h"
  #include "kudu/gutil/port.h"
  #include "kudu/gutil/stl_util.h"
  #include "kudu/gutil/stringprintf.h"
  #include "kudu/gutil/strings/stringpiece.h"
  #include "kudu/gutil/strings/substitute.h"
  #include "kudu/gutil/walltime.h"
  #include "kudu/rpc/periodic.h"
  #include "kudu/util/async_util.h"
  #include "kudu/util/debug/trace_event.h"
  #include "kudu/util/flag_tags.h"
  #include "kudu/util/logging.h"
  #include "kudu/util/metrics.h"
  #include "kudu/util/pb_util.h"
  #include "kudu/util/process_memory.h"
  #include "kudu/util/random.h"
  #include "kudu/util/random_util.h"
  #include "kudu/util/scoped_cleanup.h"
  #include "kudu/util/status.h"
  #include "kudu/util/thread_restrictions.h"
  #include "kudu/util/threadpool.h"
  #include "kudu/util/trace.h"
  #include "kudu/util/url-coding.h"
  
  DEFINE_int32(raft_heartbeat_interval_ms, 500,
               "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, 3.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_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_bool(follower_reject_update_consensus_requests, false,
              "Whether a follower will return an error for all UpdateConsensus() requests. "
              "Warning! This is only intended for testing.");
  TAG_FLAG(follower_reject_update_consensus_requests, unsafe);
  
  DEFINE_bool(follower_fail_all_prepare, false,
              "Whether a follower will fail preparing all ops. "
              "Warning! This is only intended for testing.");
  TAG_FLAG(follower_fail_all_prepare, unsafe);
  
  DEFINE_bool(raft_enable_pre_election, true,
              "When enabled, candidates will call a pre-election before "
              "running a real leader election.");
  TAG_FLAG(raft_enable_pre_election, experimental);
  TAG_FLAG(raft_enable_pre_election, runtime);
  
  DEFINE_bool(raft_enable_tombstoned_voting, true,
              "When enabled, tombstoned tablets may vote in elections.");
  TAG_FLAG(raft_enable_tombstoned_voting, experimental);
  TAG_FLAG(raft_enable_tombstoned_voting, runtime);
  
  // Enable improved re-replication (KUDU-1097).
  DEFINE_bool(raft_prepare_replacement_before_eviction, true,
              "When enabled, failed replicas will only be evicted after a "
              "replacement has been prepared for them.");
  TAG_FLAG(raft_prepare_replacement_before_eviction, advanced);
  TAG_FLAG(raft_prepare_replacement_before_eviction, experimental);
  
  DECLARE_int32(memory_limit_warn_threshold_percentage);
  
  // Metrics
  // ---------
  METRIC_DEFINE_counter(tablet, follower_memory_pressure_rejections,
                        "Follower Memory Pressure Rejections",
                        kudu::MetricUnit::kRequests,
                        "Number of RPC requests rejected due to "
                        "memory pressure while FOLLOWER.",
                        kudu::MetricLevel::kWarn);
  METRIC_DEFINE_gauge_int64(tablet, raft_term,
                            "Current Raft Consensus Term",
                            kudu::MetricUnit::kUnits,
                            "Current Term of the Raft Consensus algorithm. This number increments "
                            "each time a leader election is started.",
                            kudu::MetricLevel::kDebug);
  METRIC_DEFINE_gauge_int64(tablet, failed_elections_since_stable_leader,
                            "Failed Elections Since Stable Leader",
                            kudu::MetricUnit::kUnits,
                            "Number of failed elections on this node since there was a stable "
                            "leader. This number increments on each failed election and resets on "
                            "each successful one.",
                            kudu::MetricLevel::kWarn);
  METRIC_DEFINE_gauge_int64(tablet, time_since_last_leader_heartbeat,
                            "Time Since Last Leader Heartbeat",
                            kudu::MetricUnit::kMilliseconds,
                            "The time elapsed since the last heartbeat from the leader "
                            "in milliseconds. This metric is identically zero on a leader replica.",
                            kudu::MetricLevel::kDebug);
  
  
  using boost::optional;
  using google::protobuf::util::MessageDifferencer;
  using kudu::pb_util::SecureShortDebugString;
  using kudu::rpc::PeriodicTimer;
  using kudu::tserver::TabletServerErrorPB;
  using std::shared_ptr;
  using std::string;
  using std::unique_ptr;
  using std::unordered_set;
  using std::weak_ptr;
  using strings::Substitute;
  
  namespace kudu {
  namespace consensus {
  
  RaftConsensus::RaftConsensus(
      ConsensusOptions options,
      RaftPeerPB local_peer_pb,
      scoped_refptr<ConsensusMetadataManager> cmeta_manager,
      ServerContext server_ctx)
      : options_(std::move(options)),
        local_peer_pb_(std::move(local_peer_pb)),
        cmeta_manager_(DCHECK_NOTNULL(std::move(cmeta_manager))),
        server_ctx_(std::move(server_ctx)),
        state_(kNew),
        rng_(GetRandomSeed32()),
        leader_transfer_in_progress_(false),
        withhold_votes_until_(MonoTime::Min()),
        last_received_cur_leader_(MinimumOpId()),
        failed_elections_since_stable_leader_(0),
        shutdown_(false),
        update_calls_for_tests_(0) {
    DCHECK(local_peer_pb_.has_permanent_uuid());
  }
  
  Status RaftConsensus::Init() {
    DCHECK_EQ(kNew, state_) << State_Name(state_);
    RETURN_NOT_OK(cmeta_manager_->Load(options_.tablet_id, &cmeta_));
    SetStateUnlocked(kInitialized);
    return Status::OK();
  }
  
  RaftConsensus::~RaftConsensus() {
    Shutdown();
  }
  
  Status RaftConsensus::Create(ConsensusOptions options,
                               RaftPeerPB local_peer_pb,
                               scoped_refptr<ConsensusMetadataManager> cmeta_manager,
                               ServerContext server_ctx,
                               shared_ptr<RaftConsensus>* consensus_out) {
    shared_ptr<RaftConsensus> consensus(RaftConsensus::make_shared(std::move(options),
                                                                   std::move(local_peer_pb),
                                                                   std::move(cmeta_manager),
                                                                   std::move(server_ctx)));
    RETURN_NOT_OK_PREPEND(consensus->Init(), "Unable to initialize Raft consensus");
    *consensus_out = std::move(consensus);
    return Status::OK();
  }
  
  Status RaftConsensus::Start(const ConsensusBootstrapInfo& info,
                              unique_ptr<PeerProxyFactory> peer_proxy_factory,
                              scoped_refptr<log::Log> log,
                              unique_ptr<TimeManager> time_manager,
                              ConsensusRoundHandler* round_handler,
                              const scoped_refptr<MetricEntity>& metric_entity,
                              MarkDirtyCallback cb) {
    DCHECK(metric_entity);
  
    peer_proxy_factory_ = DCHECK_NOTNULL(std::move(peer_proxy_factory));
    log_ = DCHECK_NOTNULL(std::move(log));
    time_manager_ = DCHECK_NOTNULL(std::move(time_manager));
    round_handler_ = DCHECK_NOTNULL(round_handler);
    mark_dirty_clbk_ = std::move(cb);
  
    term_metric_ = metric_entity->FindOrCreateGauge(&METRIC_raft_term,
                                                    CurrentTerm(),
                                                    MergeType::kMax);
    follower_memory_pressure_rejections_ =
        metric_entity->FindOrCreateCounter(&METRIC_follower_memory_pressure_rejections);
  
    num_failed_elections_metric_ =
        metric_entity->FindOrCreateGauge(&METRIC_failed_elections_since_stable_leader,
                                         failed_elections_since_stable_leader_);
  
    METRIC_time_since_last_leader_heartbeat.InstantiateFunctionGauge(
        metric_entity, [this]() { return this->GetMillisSinceLastLeaderHeartbeat(); },
        MergeType::kMax)
        ->AutoDetach(&metric_detacher_);
  
    // 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.
    ThreadPool* raft_pool = server_ctx_.raft_pool;
    raft_pool_token_ = raft_pool->NewToken(ThreadPool::ExecutionMode::CONCURRENT);
  
    // The message queue that keeps track of which operations need to be replicated
    // where.
    //
    // Note: the message queue receives a dedicated Raft thread pool token so that
    // its submissions don't block other submissions by RaftConsensus (such as
    // heartbeat processing).
    //
    // TODO(adar): the token is SERIAL to match the previous single-thread
    // observer pool behavior, but CONCURRENT may be safe here.
    unique_ptr<PeerMessageQueue> queue(new PeerMessageQueue(
        metric_entity,
        log_,
        time_manager_.get(),
        local_peer_pb_,
        options_.tablet_id,
        raft_pool->NewToken(ThreadPool::ExecutionMode::SERIAL),
        server_ctx_.quiescing,
        info.last_id,
        info.last_committed_id));
  
    // A manager for the set of peers that actually send the operations both remotely
    // and to the local wal.
    unique_ptr<PeerManager> peer_manager(new PeerManager(options_.tablet_id,
                                                         peer_uuid(),
                                                         peer_proxy_factory_.get(),
                                                         queue.get(),
                                                         raft_pool_token_.get(),
                                                         log_));
  
    unique_ptr<PendingRounds> pending(new PendingRounds(
        LogPrefixThreadSafe(), time_manager_.get()));
  
    // Capture a weak_ptr reference into the functor so it can safely handle
    // outliving the consensus instance.
    weak_ptr<RaftConsensus> w = shared_from_this();
    PeriodicTimer::Options opts;
    opts.one_shot = true;
    failure_detector_ = PeriodicTimer::Create(
        peer_proxy_factory_->messenger(),
        [w]() {
          if (auto consensus = w.lock()) {
            consensus->ReportFailureDetected();
          }
        },
        MinimumElectionTimeout(),
        opts);
  
    transfer_period_timer_ = PeriodicTimer::Create(
        peer_proxy_factory_->messenger(),
        [w]() {
          if (auto consensus = w.lock()) {
            consensus->EndLeaderTransferPeriod();
          }
        },
        MinimumElectionTimeout(),
        opts);
  
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      CHECK_EQ(kInitialized, state_) << LogPrefixUnlocked() << "Illegal state for Start(): "
                                     << State_Name(state_);
  
      queue_ = std::move(queue);
      peer_manager_ = std::move(peer_manager);
      pending_ = std::move(pending);
  
      ClearLeaderUnlocked();
  
      // Our last persisted term can be higher than the last persisted operation
      // (i.e. if we called an election) but reverse should never happen.
      if (info.last_id.term() > CurrentTermUnlocked()) {
        return Status::Corruption(Substitute("Unable to start RaftConsensus: "
            "The last op in the WAL with id $0 has a term ($1) that is greater "
            "than the latest recorded term, which is $2",
            OpIdToString(info.last_id),
            info.last_id.term(),
            CurrentTermUnlocked()));
      }
  
      // Append any uncommitted replicate messages found during log replay to the queue.
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Replica starting. Triggering "
                                     << info.orphaned_replicates.size()
                                     << " pending ops. Active config: "
                                     << SecureShortDebugString(cmeta_->ActiveConfig());
      for (ReplicateMsg* replicate : info.orphaned_replicates) {
        ReplicateRefPtr replicate_ptr = make_scoped_refptr_replicate(new ReplicateMsg(*replicate));
        RETURN_NOT_OK(StartFollowerOpUnlocked(replicate_ptr));
      }
  
      // Set the initial committed opid for the PendingRounds only after
      // appending any uncommitted replicate messages to the queue.
      pending_->SetInitialCommittedOpId(info.last_committed_id);
  
      // 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.
      boost::optional<MonoDelta> fd_initial_delta;
      if (CurrentTermUnlocked() == 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 (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
          LOG_WITH_PREFIX_UNLOCKED(INFO) << "Consensus starting up: Expiring failure detector timer "
                                            "to make a prompt election more likely";
          fd_initial_delta = MonoDelta::FromMilliseconds(
              rng_.Uniform(FLAGS_raft_heartbeat_interval_ms));
        }
      }
  
      // Now assume non-leader replica duties.
      RETURN_NOT_OK(BecomeReplicaUnlocked(fd_initial_delta));
  
      SetStateUnlocked(kRunning);
    }
  
    if (IsSingleVoterConfig() && FLAGS_enable_leader_failure_detection) {
      LOG_WITH_PREFIX(INFO) << "Only one voter in the Raft config. Triggering election immediately";
      WARN_NOT_OK_EVERY_N_SECS(StartElection(NORMAL_ELECTION, INITIAL_SINGLE_NODE_ELECTION),
                               "Couldn't start leader election", 10);
    }
  
    // Report become visible to the Master.
    MarkDirty("RaftConsensus started");
  
    return Status::OK();
  }
  
  bool RaftConsensus::IsRunning() const {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    return state_ == kRunning;
  }
  
  Status RaftConsensus::EmulateElectionForTests() {
    TRACE_EVENT2("consensus", "RaftConsensus::EmulateElectionForTests",
                 "peer", peer_uuid(),
                 "tablet", options_.tablet_id);
  
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    RETURN_NOT_OK(CheckRunningUnlocked());
  
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Emulating election...";
  
    // Assume leadership of new term.
    RETURN_NOT_OK(HandleTermAdvanceUnlocked(CurrentTermUnlocked() + 1));
    SetLeaderUuidUnlocked(peer_uuid());
    return BecomeLeaderUnlocked();
  }
  
  namespace {
  const char* ModeString(RaftConsensus::ElectionMode mode) {
    switch (mode) {
      case RaftConsensus::NORMAL_ELECTION:
        return "leader election";
      case RaftConsensus::PRE_ELECTION:
        return "pre-election";
      case RaftConsensus::ELECT_EVEN_IF_LEADER_IS_ALIVE:
        return "forced leader election";
    }
    __builtin_unreachable(); // silence gcc warnings
  }
  string ReasonString(RaftConsensus::ElectionReason reason, StringPiece leader_uuid) {
    switch (reason) {
      case RaftConsensus::INITIAL_SINGLE_NODE_ELECTION:
        return "initial election of a single-replica configuration";
      case RaftConsensus::EXTERNAL_REQUEST:
        return "received explicit request";
      case RaftConsensus::ELECTION_TIMEOUT_EXPIRED:
        if (leader_uuid.empty()) {
          return "no leader contacted us within the election timeout";
        }
        return Substitute("detected failure of leader $0", leader_uuid);
    }
    __builtin_unreachable(); // silence gcc warnings
  }
  } // anonymous namespace
  
  Status RaftConsensus::StartElection(ElectionMode mode, ElectionReason reason) {
    if (server_ctx_.quiescing && server_ctx_.quiescing->load()) {
      return Status::IllegalState("leader elections are disabled");
    }
    const char* const mode_str = ModeString(mode);
  
    TRACE_EVENT2("consensus", "RaftConsensus::StartElection",
                 "peer", LogPrefixThreadSafe(),
                 "mode", mode_str);
    scoped_refptr<LeaderElection> election;
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      RETURN_NOT_OK(CheckRunningUnlocked());
  
      RaftPeerPB::Role active_role = cmeta_->active_role();
      if (active_role == RaftPeerPB::LEADER) {
        LOG_WITH_PREFIX_UNLOCKED(INFO) << Substitute(
            "Not starting $0 -- already a leader", mode_str);
        return Status::OK();
      }
      if (PREDICT_FALSE(!consensus::IsVoterRole(active_role))) {
        // A non-voter should not start leader elections. The leader failure
        // detector should be re-enabled once the non-voter replica is promoted
        // to voter replica.
        return Status::IllegalState("only voting members can start elections",
            SecureShortDebugString(cmeta_->ActiveConfig()));
      }
      if (PREDICT_FALSE(active_role == RaftPeerPB::NON_PARTICIPANT)) {
        SnoozeFailureDetector();
        return Status::IllegalState("Not starting election: node is currently "
                                    "a non-participant in the Raft config",
                                    SecureShortDebugString(cmeta_->ActiveConfig()));
      }
      LOG_WITH_PREFIX_UNLOCKED(INFO)
          << "Starting " << mode_str
          << " (" << ReasonString(reason, GetLeaderUuidUnlocked()) << ")";
  
      // Snooze to avoid the election timer firing again as much as possible.
      // We do not disable the election timer while running an election, so that
      // if the election times out, we will try again.
      MonoDelta timeout = LeaderElectionExpBackoffDeltaUnlocked();
      SnoozeFailureDetector(string("starting election"), timeout);
  
      // Increment the term and vote for ourselves, unless it's a pre-election.
      if (mode != PRE_ELECTION) {
        // TODO(mpercy): Consider using a separate Mutex for voting, which must sync to disk.
  
        // We skip flushing the term to disk because setting the vote just below also
        // flushes to disk, and the double fsync doesn't buy us anything.
        RETURN_NOT_OK(HandleTermAdvanceUnlocked(CurrentTermUnlocked() + 1,
                                                SKIP_FLUSH_TO_DISK));
        RETURN_NOT_OK(SetVotedForCurrentTermUnlocked(peer_uuid()));
      }
  
      RaftConfigPB active_config = cmeta_->ActiveConfig();
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Starting " << mode_str << " with config: "
                                     << SecureShortDebugString(active_config);
  
      // Initialize the VoteCounter.
      int num_voters = CountVoters(active_config);
      int majority_size = MajoritySize(num_voters);
      VoteCounter counter(num_voters, majority_size);
  
      // Vote for ourselves.
      bool duplicate;
      RETURN_NOT_OK(counter.RegisterVote(peer_uuid(), VOTE_GRANTED, &duplicate));
      CHECK(!duplicate) << LogPrefixUnlocked()
                        << "Inexplicable duplicate self-vote for term "
                        << CurrentTermUnlocked();
  
      VoteRequestPB request;
      request.set_ignore_live_leader(mode == ELECT_EVEN_IF_LEADER_IS_ALIVE);
      request.set_candidate_uuid(peer_uuid());
      if (mode == PRE_ELECTION) {
        // In a pre-election, we haven't bumped our own term yet, so we need to be
        // asking for votes for the next term.
        request.set_is_pre_election(true);
        request.set_candidate_term(CurrentTermUnlocked() + 1);
      } else {
        request.set_candidate_term(CurrentTermUnlocked());
      }
      request.set_tablet_id(options_.tablet_id);
      *request.mutable_candidate_status()->mutable_last_received() =
          queue_->GetLastOpIdInLog();
  
      auto self = shared_from_this();
      election.reset(new LeaderElection(
          std::move(active_config),
          // The RaftConsensus ref passed below ensures that this raw pointer
          // remains safe to use for the entirety of LeaderElection's life.
          peer_proxy_factory_.get(),
          std::move(request), std::move(counter), timeout,
          [self, reason](const ElectionResult& result) {
            self->ElectionCallback(reason, result);
          }));
    }
  
    // Start the election outside the lock.
    election->Run();
  
    return Status::OK();
  }
  
  Status RaftConsensus::WaitUntilLeaderForTests(const MonoDelta& timeout) {
    MonoTime deadline = MonoTime::Now() + timeout;
    while (role() != consensus::RaftPeerPB::LEADER) {
      if (MonoTime::Now() >= deadline) {
        return Status::TimedOut(Substitute("Peer $0 is not leader of tablet $1 after $2. Role: $3",
                                           peer_uuid(), options_.tablet_id, timeout.ToString(),
                                           role()));
      }
      SleepFor(MonoDelta::FromMilliseconds(10));
    }
    return Status::OK();
  }
  
  Status RaftConsensus::StepDown(LeaderStepDownResponsePB* resp) {
    TRACE_EVENT0("consensus", "RaftConsensus::StepDown");
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    DCHECK((queue_->IsInLeaderMode() && cmeta_->active_role() == RaftPeerPB::LEADER) ||
           (!queue_->IsInLeaderMode() && cmeta_->active_role() != RaftPeerPB::LEADER));
    RETURN_NOT_OK(CheckRunningUnlocked());
    if (cmeta_->active_role() != RaftPeerPB::LEADER) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Rejecting request to step down while not 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();
    }
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Received request to step down";
    RETURN_NOT_OK(HandleTermAdvanceUnlocked(CurrentTermUnlocked() + 1,
                                            SKIP_FLUSH_TO_DISK));
    // Snooze the failure detector for an extra leader failure timeout.
    // This should ensure that a different replica is elected leader after this one steps down.
    SnoozeFailureDetector(string("explicit stepdown request"), MonoDelta::FromMilliseconds(
        2 * MinimumElectionTimeout().ToMilliseconds()));
    return Status::OK();
  }
  
  Status RaftConsensus::TransferLeadership(const boost::optional<string>& new_leader_uuid,
                                           LeaderStepDownResponsePB* resp) {
    TRACE_EVENT0("consensus", "RaftConsensus::TransferLeadership");
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Received request to transfer leadership"
                                   << (new_leader_uuid ?
                                      Substitute(" to TS $0", *new_leader_uuid) :
                                      "");
    DCHECK((queue_->IsInLeaderMode() && cmeta_->active_role() == RaftPeerPB::LEADER) ||
           (!queue_->IsInLeaderMode() && cmeta_->active_role() != RaftPeerPB::LEADER));
    RETURN_NOT_OK(CheckRunningUnlocked());
    if (cmeta_->active_role() != RaftPeerPB::LEADER) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Rejecting request to transer leadership while not 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();
    }
    if (new_leader_uuid) {
      if (*new_leader_uuid == peer_uuid()) {
        // Short-circuit as we are transferring leadership to ourselves and we
        // already checked that we are leader.
        return Status::OK();
      }
      if (!IsRaftConfigVoter(*new_leader_uuid, cmeta_->ActiveConfig())) {
        const string msg = Substitute("tablet server $0 is not a voter in the active config",
                                      *new_leader_uuid);
        LOG_WITH_PREFIX_UNLOCKED(INFO) << "Rejecting request to transfer leadership "
                                       << "because " << msg;
        return Status::InvalidArgument(msg);
      }
    }
    return BeginLeaderTransferPeriodUnlocked(new_leader_uuid);
  }
  
  Status RaftConsensus::BeginLeaderTransferPeriodUnlocked(
      const boost::optional<string>& successor_uuid) {
    DCHECK(lock_.is_locked());
    if (leader_transfer_in_progress_.CompareAndSwap(false, true)) {
      return Status::ServiceUnavailable(
          Substitute("leadership transfer for $0 already in progress",
                     options_.tablet_id));
    }
    leader_transfer_in_progress_.Store(true, kMemOrderAcquire);
    queue_->BeginWatchForSuccessor(successor_uuid);
    transfer_period_timer_->Start();
    return Status::OK();
  }
  
  void RaftConsensus::EndLeaderTransferPeriod() {
    transfer_period_timer_->Stop();
    queue_->EndWatchForSuccessor();
    leader_transfer_in_progress_.Store(false, kMemOrderRelease);
  }
  
  scoped_refptr<ConsensusRound> RaftConsensus::NewRound(
      unique_ptr<ReplicateMsg> replicate_msg,
      ConsensusReplicatedCallback replicated_cb) {
    return make_scoped_refptr(new ConsensusRound(this,
                                                 std::move(replicate_msg),
                                                 std::move(replicated_cb)));
  }
  
  void RaftConsensus::ReportFailureDetectedTask() {
    Status s = StartElection(FLAGS_raft_enable_pre_election ?
        PRE_ELECTION : NORMAL_ELECTION, ELECTION_TIMEOUT_EXPIRED);
    if (PREDICT_FALSE(!s.ok())) {
      WARN_NOT_OK_EVERY_N_SECS(
          s, LogPrefixThreadSafe() + "failed to trigger leader election", 10);
      // Normally the failure detector would be enabled at the end of the election,
      // but since the election failed to start, we must reenable explicitly.
      EnableFailureDetector();
    }
  }
  
  void RaftConsensus::ReportFailureDetected() {
    // We're running on a timer thread; start an election on a different thread.
    //
    // There's no need to reenable the failure detector; if this fails, it's a
    // sign that RaftConsensus has stopped and we no longer need failure detection.
    auto self = shared_from_this();
    Status s = raft_pool_token_->Submit([self]() { self->ReportFailureDetectedTask(); });
    if (PREDICT_FALSE(!s.ok())) {
      static const char* msg = "failed to submit failure detected task";
      CHECK(s.IsServiceUnavailable()) << LogPrefixThreadSafe() << msg;
      WARN_NOT_OK(s, LogPrefixThreadSafe() + msg);
    }
  }
  
  Status RaftConsensus::BecomeLeaderUnlocked() {
    DCHECK(lock_.is_locked());
  
    TRACE_EVENT2("consensus", "RaftConsensus::BecomeLeaderUnlocked",
                 "peer", peer_uuid(),
                 "tablet", options_.tablet_id);
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Becoming Leader. State: " << ToStringUnlocked();
  
    // Disable FD while we are leader.
    DisableFailureDetector();
  
    // Don't vote for anyone if we're a leader.
    withhold_votes_until_ = MonoTime::Max();
  
    // Leadership never starts in a transfer period.
    EndLeaderTransferPeriod();
  
    queue_->RegisterObserver(this);
    bool was_leader = queue_->IsInLeaderMode();
    RETURN_NOT_OK(RefreshConsensusQueueAndPeersUnlocked());
    if (!was_leader && server_ctx_.num_leaders) server_ctx_.num_leaders->Increment();
  
    // Initiate a NO_OP op that is sent at the beginning of every term
    // change in raft.
    auto replicate = new ReplicateMsg;
    replicate->set_op_type(NO_OP);
    replicate->mutable_noop_request(); // Define the no-op request field.
    CHECK_OK(time_manager_->AssignTimestamp(replicate));
  
    scoped_refptr<ConsensusRound> round(
        new ConsensusRound(this, make_scoped_refptr(new RefCountedReplicate(replicate))));
    auto* round_raw = round.get();
    round->SetConsensusReplicatedCallback(
        [this, round_raw](const Status& s) {
          this->NonTxRoundReplicationFinished(round_raw, &DoNothingStatusCB, s);
        });
  
    last_leader_communication_time_micros_ = 0;
  
    return AppendNewRoundToQueueUnlocked(round);
  }
  
  Status RaftConsensus::BecomeReplicaUnlocked(boost::optional<MonoDelta> fd_delta) {
    DCHECK(lock_.is_locked());
  
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Becoming Follower/Learner. State: "
                                   << ToStringUnlocked();
    ClearLeaderUnlocked();
  
    // Enable/disable leader failure detection if becoming VOTER/NON_VOTER replica
    // correspondingly.
    UpdateFailureDetectorState(std::move(fd_delta));
  
    // Now that we're a replica, we can allow voting for other nodes.
    withhold_votes_until_ = MonoTime::Min();
  
    // Deregister ourselves from the queue. We no longer need to track what gets
    // replicated since we're stepping down.
    queue_->UnRegisterObserver(this);
    bool was_leader = queue_->IsInLeaderMode();
    queue_->SetNonLeaderMode(cmeta_->ActiveConfig());
    if (was_leader && server_ctx_.num_leaders) {
      server_ctx_.num_leaders->IncrementBy(-1);
    }
    peer_manager_->Close();
  
    return Status::OK();
  }
  
  Status RaftConsensus::Replicate(const scoped_refptr<ConsensusRound>& round) {
  
    std::lock_guard<simple_spinlock> lock(update_lock_);
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      RETURN_NOT_OK(CheckSafeToReplicateUnlocked(*round->replicate_msg()));
      RETURN_NOT_OK(round->CheckBoundTerm(CurrentTermUnlocked()));
      RETURN_NOT_OK(AppendNewRoundToQueueUnlocked(round));
    }
  
    peer_manager_->SignalRequest();
    return Status::OK();
  }
  
  Status RaftConsensus::CheckLeadershipAndBindTerm(const scoped_refptr<ConsensusRound>& round) {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    RETURN_NOT_OK(CheckSafeToReplicateUnlocked(*round->replicate_msg()));
    round->BindToTerm(CurrentTermUnlocked());
    return Status::OK();
  }
  
  Status RaftConsensus::AppendNewRoundToQueueUnlocked(const scoped_refptr<ConsensusRound>& round) {
    DCHECK(lock_.is_locked());
  
    *round->replicate_msg()->mutable_id() = queue_->GetNextOpId();
    RETURN_NOT_OK(AddPendingOperationUnlocked(round));
  
    // The only reasons for a bad status would be if the log itself were shut down,
    // or if we had an actual IO error, which we currently don't handle.
    CHECK_OK_PREPEND(queue_->AppendOperation(round->replicate_scoped_refptr()),
                     Substitute("$0: could not append to queue", LogPrefixUnlocked()));
    return Status::OK();
  }
  
  Status RaftConsensus::AddPendingOperationUnlocked(const scoped_refptr<ConsensusRound>& round) {
    DCHECK(lock_.is_locked());
    DCHECK(pending_);
  
    // If we are adding a pending config change, we need to propagate it to the
    // metadata.
    if (PREDICT_FALSE(round->replicate_msg()->op_type() == CHANGE_CONFIG_OP)) {
      // Fill in the opid for the proposed new configuration. This has to be done
      // here rather than when it's first created because we didn't yet have an
      // OpId assigned at creation time.
      ChangeConfigRecordPB* change_record = round->replicate_msg()->mutable_change_config_record();
      change_record->mutable_new_config()->set_opid_index(round->replicate_msg()->id().index());
  
      DCHECK(change_record->IsInitialized())
          << "change_config_record missing required fields: "
          << change_record->InitializationErrorString();
  
      const RaftConfigPB& new_config = change_record->new_config();
  
      if (!new_config.unsafe_config_change()) {
        Status s = CheckNoConfigChangePendingUnlocked();
        if (PREDICT_FALSE(!s.ok())) {
          s = s.CloneAndAppend(Substitute("\n  New config: $0", SecureShortDebugString(new_config)));
          LOG_WITH_PREFIX_UNLOCKED(INFO) << s.ToString();
          return s;
        }
      }
      // Check if the pending Raft config has an OpId less than the committed
      // config. If so, this is a replay at startup in which the COMMIT
      // messages were delayed.
      int64_t committed_config_opid_index = cmeta_->GetConfigOpIdIndex(COMMITTED_CONFIG);
      if (round->replicate_msg()->id().index() > committed_config_opid_index) {
        RETURN_NOT_OK(SetPendingConfigUnlocked(new_config));
        if (cmeta_->active_role() == RaftPeerPB::LEADER) {
          RETURN_NOT_OK(RefreshConsensusQueueAndPeersUnlocked());
        }
      } else {
        LOG_WITH_PREFIX_UNLOCKED(INFO)
            << "Ignoring setting pending config change with OpId "
            << round->replicate_msg()->id() << " because the committed config has OpId index "
            << committed_config_opid_index << ". The config change we are ignoring is: "
            << "Old config: { " << SecureShortDebugString(change_record->old_config()) << " }. "
            << "New config: { " << SecureShortDebugString(new_config) << " }";
      }
    }
  
    return pending_->AddPendingOperation(round);
  }
  
  void RaftConsensus::NotifyCommitIndex(int64_t commit_index) {
    TRACE_EVENT2("consensus", "RaftConsensus::NotifyCommitIndex",
                 "tablet", options_.tablet_id,
                 "commit_index", commit_index);
  
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    // We will process commit notifications while shutting down because a replica
    // which has initiated a Prepare() / Replicate() may eventually commit even if
    // its state has changed after the initial Append() / Update().
    if (PREDICT_FALSE(state_ != kRunning && state_ != kStopping)) {
      LOG_WITH_PREFIX_UNLOCKED(WARNING) << "Unable to update committed index: "
                                        << "Replica not in running state: "
                                        << State_Name(state_);
      return;
    }
  
    pending_->AdvanceCommittedIndex(commit_index);
  
    if (cmeta_->active_role() == RaftPeerPB::LEADER) {
      peer_manager_->SignalRequest(false);
    }
  }
  
  void RaftConsensus::NotifyTermChange(int64_t term) {
    TRACE_EVENT2("consensus", "RaftConsensus::NotifyTermChange",
                 "tablet", options_.tablet_id,
                 "term", term);
  
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    Status s = CheckRunningUnlocked();
    if (PREDICT_FALSE(!s.ok())) {
      LOG_WITH_PREFIX_UNLOCKED(WARNING) << "Unable to handle notification of new term "
                                        << "(" << term << "): " << s.ToString();
      return;
    }
    WARN_NOT_OK(HandleTermAdvanceUnlocked(term), "Couldn't advance consensus term.");
  }
  
  void RaftConsensus::NotifyFailedFollower(const string& uuid,
                                           int64_t term,
                                           const string& reason) {
    // Common info used in all of the log messages within this method.
    string fail_msg = Substitute("Processing failure of peer $0 in term $1 ($2): ",
                                 uuid, term, reason);
  
    if (!FLAGS_evict_failed_followers) {
      LOG(INFO) << LogPrefixThreadSafe() << fail_msg
                << "Eviction of failed followers is disabled. Doing nothing.";
      return;
    }
  
    RaftConfigPB committed_config;
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      int64_t current_term = CurrentTermUnlocked();
      if (current_term != term) {
        LOG_WITH_PREFIX_UNLOCKED(INFO) << fail_msg << "Notified about a follower failure in "
                                       << "previous term " << term << ", but a leader election "
                                       << "likely occurred since the failure was detected. "
                                       << "Doing nothing.";
        return;
      }
  
      if (cmeta_->has_pending_config()) {
        LOG_WITH_PREFIX_UNLOCKED(INFO) << fail_msg << "There is already a config change operation "
                                       << "in progress. Unable to evict follower until it completes. "
                                       << "Doing nothing.";
        return;
      }
      committed_config = cmeta_->CommittedConfig();
    }
  
    // Run config change on thread pool after dropping lock.
    auto self = shared_from_this();
    WARN_NOT_OK(raft_pool_token_->Submit(
        [self, uuid, committed_config, reason]() {
          self->TryRemoveFollowerTask(uuid, committed_config, reason);
        }),
      LogPrefixThreadSafe() + "Unable to start TryRemoveFollowerTask");
  }
  
  void RaftConsensus::NotifyPeerToPromote(const string& peer_uuid) {
    // Run the config change on the raft thread pool.
    auto self = shared_from_this();
    WARN_NOT_OK(raft_pool_token_->Submit(
        [self, peer_uuid]() { self->TryPromoteNonVoterTask(peer_uuid); }),
      LogPrefixThreadSafe() + "Unable to start TryPromoteNonVoterTask");
  }
  
  void RaftConsensus::NotifyPeerToStartElection(const string& peer_uuid) {
    const auto& log_prefix = LogPrefixThreadSafe();
    LOG(INFO) << log_prefix << ": Instructing follower " << peer_uuid << " to start an election";
    auto self = shared_from_this();
    WARN_NOT_OK(raft_pool_token_->Submit(
        [self, peer_uuid]() { self->TryStartElectionOnPeerTask(peer_uuid); }),
      log_prefix + "Unable to start TryStartElectionOnPeerTask");
  }
  
  void RaftConsensus::NotifyPeerHealthChange() {
    MarkDirty("Peer health change");
  }
  
  void RaftConsensus::TryRemoveFollowerTask(const string& uuid,
                                            const RaftConfigPB& committed_config,
                                            const string& reason) {
    ChangeConfigRequestPB req;
    req.set_tablet_id(options_.tablet_id);
    req.mutable_server()->set_permanent_uuid(uuid);
    req.set_type(REMOVE_PEER);
    req.set_cas_config_opid_index(committed_config.opid_index());
    LOG(INFO) << LogPrefixThreadSafe() << "Attempting to remove follower "
              << uuid << " from the Raft config. Reason: " << reason;
    boost::optional<TabletServerErrorPB::Code> error_code;
    WARN_NOT_OK(ChangeConfig(req, &DoNothingStatusCB, &error_code),
                LogPrefixThreadSafe() + "Unable to remove follower " + uuid);
  }
  
  void RaftConsensus::TryPromoteNonVoterTask(const string& peer_uuid) {
    string msg = Substitute("attempt to promote peer $0: ", peer_uuid);
    int64_t current_committed_config_index;
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
  
      if (cmeta_->has_pending_config()) {
       LOG_WITH_PREFIX_UNLOCKED(INFO) << msg << "there is already a config change operation "
                                       << "in progress. Unable to promote follower until it "
                                       << "completes. Doing nothing.";
        return;
      }
  
      // Check if the peer is still part of the current committed config.
      RaftConfigPB committed_config = cmeta_->CommittedConfig();
      current_committed_config_index = committed_config.opid_index();
  
      RaftPeerPB* peer_pb;
      Status s = GetRaftConfigMember(&committed_config, peer_uuid, &peer_pb);
      if (!s.ok()) {
        LOG_WITH_PREFIX_UNLOCKED(INFO) << msg << "can't find peer in the "
                                       << "current committed config: "
                                       << committed_config.ShortDebugString()
                                       << ". Doing nothing.";
        return;
      }
  
      // Also check if the peer it still a NON_VOTER waiting for promotion.
      if (peer_pb->member_type() != RaftPeerPB::NON_VOTER || !peer_pb->attrs().promote()) {
        LOG_WITH_PREFIX_UNLOCKED(INFO) << msg << "peer is either no longer a NON_VOTER "
                                       << "or not marked for promotion anymore. Current "
                                       << "config: " << committed_config.ShortDebugString()
                                       << ". Doing nothing.";
        return;
      }
    }
  
    ChangeConfigRequestPB req;
    req.set_tablet_id(options_.tablet_id);
    req.set_type(MODIFY_PEER);
    req.mutable_server()->set_permanent_uuid(peer_uuid);
    req.mutable_server()->set_member_type(RaftPeerPB::VOTER);
    req.mutable_server()->mutable_attrs()->set_promote(false);
    req.set_cas_config_opid_index(current_committed_config_index);
    LOG(INFO) << LogPrefixThreadSafe() << "attempting to promote NON_VOTER "
              << peer_uuid << " to VOTER";
    boost::optional<TabletServerErrorPB::Code> error_code;
    WARN_NOT_OK(ChangeConfig(req, &DoNothingStatusCB, &error_code),
                LogPrefixThreadSafe() + Substitute("Unable to promote non-voter $0", peer_uuid));
  }
  
  void RaftConsensus::TryStartElectionOnPeerTask(const string& peer_uuid) {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    // Double-check that the peer is a voter in the active config.
    if (!IsRaftConfigVoter(peer_uuid, cmeta_->ActiveConfig())) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Not signalling peer " << peer_uuid
                                     << "to start an election: it's not a voter "
                                     << "in the active config.";
      return;
    }
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Signalling peer " << peer_uuid
                                   << " to start an election";
    WARN_NOT_OK(peer_manager_->StartElection(peer_uuid),
                Substitute("unable to start election on peer $0", peer_uuid));
  }
  
  Status RaftConsensus::Update(const ConsensusRequestPB* request,
                               ConsensusResponsePB* response) {
    update_calls_for_tests_.Increment();
  
    if (PREDICT_FALSE(FLAGS_follower_reject_update_consensus_requests)) {
      return Status::IllegalState("Rejected: --follower_reject_update_consensus_requests "
                                  "is set to true.");
    }
  
    response->set_responder_uuid(peer_uuid());
  
    VLOG_WITH_PREFIX(2) << "Replica received request: " << SecureShortDebugString(*request);
  
    // see var declaration
    std::lock_guard<simple_spinlock> lock(update_lock_);
    Status s = UpdateReplica(request, response);
    if (PREDICT_FALSE(VLOG_IS_ON(1))) {
      if (request->ops().empty()) {
        VLOG_WITH_PREFIX(1)
            << Substitute("Replica replied to status only request. Replica: $0. Response: $1",
                          ToString(), SecureShortDebugString(*response));
      }
    }
    return s;
  }
  
  // Helper function to check if the op is a no-op op.
  static bool IsConsensusOnlyOperation(OperationType op_type) {
    return op_type == NO_OP || op_type == CHANGE_CONFIG_OP;
  }
  
  Status RaftConsensus::StartFollowerOpUnlocked(const ReplicateRefPtr& msg) {
    DCHECK(lock_.is_locked());
  
    if (IsConsensusOnlyOperation(msg->get()->op_type())) {
      return StartConsensusOnlyRoundUnlocked(msg);
    }
  
    if (PREDICT_FALSE(FLAGS_follower_fail_all_prepare)) {
      return Status::IllegalState("Rejected: --follower_fail_all_prepare "
                                  "is set to true.");
    }
  
    VLOG_WITH_PREFIX_UNLOCKED(1) << "Starting op: "
                                 << SecureShortDebugString(msg->get()->id());
    scoped_refptr<ConsensusRound> round(new ConsensusRound(this, msg));
    ConsensusRound* round_ptr = round.get();
    RETURN_NOT_OK(round_handler_->StartFollowerOp(round));
    return AddPendingOperationUnlocked(round_ptr);
  }
  
  bool RaftConsensus::IsSingleVoterConfig() const {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    return cmeta_->CountVotersInConfig(COMMITTED_CONFIG) == 1 &&
           cmeta_->IsVoterInConfig(peer_uuid(), COMMITTED_CONFIG);
  }
  
  string RaftConsensus::LeaderRequest::OpsRangeString() const {
    string ret;
    ret.reserve(100);
    ret.push_back('[');
    if (!messages.empty()) {
      const OpId& first_op = (*messages.begin())->get()->id();
      const OpId& last_op = (*messages.rbegin())->get()->id();
      strings::SubstituteAndAppend(&ret, "$0.$1-$2.$3",
                                   first_op.term(), first_op.index(),
                                   last_op.term(), last_op.index());
    }
    ret.push_back(']');
    return ret;
  }
  
  void RaftConsensus::DeduplicateLeaderRequestUnlocked(ConsensusRequestPB* rpc_req,
                                                       LeaderRequest* deduplicated_req) {
    DCHECK(lock_.is_locked());
  
    // TODO(todd): use queue committed index?
    int64_t last_committed_index = pending_->GetCommittedIndex();
  
    // The leader's preceding id.
    deduplicated_req->preceding_opid = &rpc_req->preceding_id();
  
    int64_t dedup_up_to_index = queue_->GetLastOpIdInLog().index();
  
    deduplicated_req->first_message_idx = -1;
  
    // In this loop we discard duplicates and advance the leader's preceding id
    // accordingly.
    for (int i = 0; i < rpc_req->ops_size(); i++) {
      ReplicateMsg* leader_msg = rpc_req->mutable_ops(i);
  
      if (leader_msg->id().index() <= last_committed_index) {
        VLOG_WITH_PREFIX_UNLOCKED(2) << "Skipping op id " << leader_msg->id()
                                     << " (already committed)";
        deduplicated_req->preceding_opid = &leader_msg->id();
        continue;
      }
  
      if (leader_msg->id().index() <= dedup_up_to_index) {
        // If the index is uncommitted and below our match index, then it must be in the
        // pendings set.
        scoped_refptr<ConsensusRound> round =
            pending_->GetPendingOpByIndexOrNull(leader_msg->id().index());
        DCHECK(round) << "Could not find op with index " << leader_msg->id().index()
                      << " in pending set. committed= " << last_committed_index
                      << " dedup=" << dedup_up_to_index;
  
        // If the OpIds match, i.e. if they have the same term and id, then this is just
        // duplicate, we skip...
        if (OpIdEquals(round->replicate_msg()->id(), leader_msg->id())) {
          VLOG_WITH_PREFIX_UNLOCKED(2) << "Skipping op id " << leader_msg->id()
                                       << " (already replicated)";
          deduplicated_req->preceding_opid = &leader_msg->id();
          continue;
        }
  
        // ... otherwise we must adjust our match index, i.e. all messages from now on
        // are "new"
        dedup_up_to_index = leader_msg->id().index();
      }
  
      if (deduplicated_req->first_message_idx == - 1) {
        deduplicated_req->first_message_idx = i;
      }
      deduplicated_req->messages.push_back(make_scoped_refptr_replicate(leader_msg));
    }
  
    if (deduplicated_req->messages.size() != rpc_req->ops_size()) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Deduplicated request from leader. Original: "
                            << rpc_req->preceding_id() << "->" << OpsRangeString(*rpc_req)
                            << "   Dedup: " << *deduplicated_req->preceding_opid << "->"
                            << deduplicated_req->OpsRangeString();
    }
  
  }
  
  Status RaftConsensus::HandleLeaderRequestTermUnlocked(const ConsensusRequestPB* request,
                                                        ConsensusResponsePB* response) {
    DCHECK(lock_.is_locked());
    // Do term checks first:
    if (PREDICT_FALSE(request->caller_term() != CurrentTermUnlocked())) {
  
      // If less, reject.
      if (request->caller_term() < CurrentTermUnlocked()) {
        string msg = Substitute("Rejecting Update request from peer $0 for earlier term $1. "
                                "Current term is $2. Ops: $3",
  
                                request->caller_uuid(),
                                request->caller_term(),
                                CurrentTermUnlocked(),
                                OpsRangeString(*request));
        LOG_WITH_PREFIX_UNLOCKED(INFO) << msg;
        FillConsensusResponseError(response,
                                   ConsensusErrorPB::INVALID_TERM,
                                   Status::IllegalState(msg));
        return Status::OK();
      }
      RETURN_NOT_OK(HandleTermAdvanceUnlocked(request->caller_term()));
    }
    return Status::OK();
  }
  
  Status RaftConsensus::EnforceLogMatchingPropertyMatchesUnlocked(const LeaderRequest& req,
                                                                  ConsensusResponsePB* response) {
    DCHECK(lock_.is_locked());
  
    bool term_mismatch;
    if (pending_->IsOpCommittedOrPending(*req.preceding_opid, &term_mismatch)) {
      return Status::OK();
    }
  
    string error_msg = Substitute(
      "Log matching property violated."
      " Preceding OpId in replica: $0. Preceding OpId from leader: $1. ($2 mismatch)",
      SecureShortDebugString(queue_->GetLastOpIdInLog()),
      SecureShortDebugString(*req.preceding_opid),
      term_mismatch ? "term" : "index");
  
  
    FillConsensusResponseError(response,
                               ConsensusErrorPB::PRECEDING_ENTRY_DIDNT_MATCH,
                               Status::IllegalState(error_msg));
  
    // Adding a check to eliminate an unnecessary log message in the
    // scenario where this is the first message from the Leader of a new tablet.
    if (!OpIdEquals(MakeOpId(1,1), *req.preceding_opid)) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Refusing update from remote peer "
                                     << req.leader_uuid << ": " << error_msg;
    }
  
    // If the terms mismatch we abort down to the index before the leader's preceding,
    // since we know that is the last opid that has a chance of not being overwritten.
    // Aborting preemptively here avoids us reporting a last received index that is
    // possibly higher than the leader's causing an avoidable cache miss on the leader's
    // queue.
    //
    // TODO: this isn't just an optimization! if we comment this out, we get
    // failures on raft_consensus-itest a couple percent of the time! Should investigate
    // why this is actually critical to do here, as opposed to just on requests that
    // append some ops.
    if (term_mismatch) {
      TruncateAndAbortOpsAfterUnlocked(req.preceding_opid->index() - 1);
    }
  
    return Status::OK();
  }
  
  void RaftConsensus::TruncateAndAbortOpsAfterUnlocked(int64_t truncate_after_index) {
    DCHECK(lock_.is_locked());
    pending_->AbortOpsAfter(truncate_after_index);
    queue_->TruncateOpsAfter(truncate_after_index);
  }
  
  Status RaftConsensus::CheckLeaderRequestUnlocked(const ConsensusRequestPB* request,
                                                   ConsensusResponsePB* response,
                                                   LeaderRequest* deduped_req) {
    DCHECK(lock_.is_locked());
  
    if (request->has_deprecated_committed_index() ||
        !request->has_all_replicated_index()) {
      return Status::InvalidArgument("Leader appears to be running an earlier version "
                                     "of Kudu. Please shut down and upgrade all servers "
                                     "before restarting.");
    }
  
    ConsensusRequestPB* mutable_req = const_cast<ConsensusRequestPB*>(request);
    DeduplicateLeaderRequestUnlocked(mutable_req, deduped_req);
  
    // This is an additional check for KUDU-639 that makes sure the message's index
    // and term are in the right sequence in the request, after we've deduplicated
    // them. We do this before we change any of the internal state.
    //
    // TODO move this to raft_consensus-state or whatever we transform that into.
    // We should be able to do this check for each append, but right now the way
    // we initialize raft_consensus-state is preventing us from doing so.
    Status s;
    const OpId* prev = deduped_req->preceding_opid;
    for (const ReplicateRefPtr& message : deduped_req->messages) {
      s = PendingRounds::CheckOpInSequence(*prev, message->get()->id());
      if (PREDICT_FALSE(!s.ok())) {
        LOG(ERROR) << "Leader request contained out-of-sequence messages. Status: "
            << s.ToString() << ". Leader Request: " << SecureShortDebugString(*request);
        break;
      }
      prev = &message->get()->id();
    }
  
    // We only release the messages from the request after the above check so that
    // that we can print the original request, if it fails.
    if (!deduped_req->messages.empty()) {
      // We take ownership of the deduped ops.
      DCHECK_GE(deduped_req->first_message_idx, 0);
      mutable_req->mutable_ops()->ExtractSubrange(
          deduped_req->first_message_idx,
          deduped_req->messages.size(),
          nullptr);
    }
  
    RETURN_NOT_OK(s);
  
    RETURN_NOT_OK(HandleLeaderRequestTermUnlocked(request, response));
  
    if (response->status().has_error()) {
      return Status::OK();
    }
  
    RETURN_NOT_OK(EnforceLogMatchingPropertyMatchesUnlocked(*deduped_req, response));
  
    if (response->status().has_error()) {
      return Status::OK();
    }
  
    // If the first of the messages to apply is not in our log, either it follows the last
    // received message or it replaces some in-flight.
    if (!deduped_req->messages.empty()) {
  
      bool term_mismatch;
      CHECK(!pending_->IsOpCommittedOrPending(deduped_req->messages[0]->get()->id(), &term_mismatch));
  
      // If the index is in our log but the terms are not the same abort down to the leader's
      // preceding id.
      if (term_mismatch) {
        TruncateAndAbortOpsAfterUnlocked(deduped_req->preceding_opid->index());
      }
    }
  
    // If all of the above logic was successful then we can consider this to be
    // the effective leader of the configuration. If they are not currently marked as
    // the leader locally, mark them as leader now.
    const string& caller_uuid = request->caller_uuid();
    if (PREDICT_FALSE(HasLeaderUnlocked() &&
                      GetLeaderUuidUnlocked() != caller_uuid)) {
      LOG_WITH_PREFIX_UNLOCKED(FATAL) << "Unexpected new leader in same term! "
          << "Existing leader UUID: " << GetLeaderUuidUnlocked() << ", "
          << "new leader UUID: " << caller_uuid;
    }
    if (PREDICT_FALSE(!HasLeaderUnlocked())) {
      SetLeaderUuidUnlocked(request->caller_uuid());
    }
  
    return Status::OK();
  }
  
  Status RaftConsensus::UpdateReplica(const ConsensusRequestPB* request,
                                      ConsensusResponsePB* response) {
    TRACE_EVENT2("consensus", "RaftConsensus::UpdateReplica",
                 "peer", peer_uuid(),
                 "tablet", options_.tablet_id);
    Synchronizer log_synchronizer;
    StatusCallback sync_status_cb = log_synchronizer.AsStatusCallback();
  
  
    // The ordering of the following operations is crucial, read on for details.
    //
    // The main requirements explained in more detail below are:
    //
    //   1) We must enqueue the prepares before we write to our local log.
    //   2) If we were able to enqueue a prepare then we must be able to log it.
    //   3) If we fail to enqueue a prepare, we must not attempt to enqueue any
    //      later-indexed prepare or apply.
    //
    // See below for detailed rationale.
    //
    // The steps are:
    //
    // 0 - Split/Dedup
    //
    // We split the operations into replicates and commits and make sure that we don't
    // don't do anything on operations we've already received in a previous call.
    // This essentially makes this method idempotent.
    //
    // 1 - We mark as many pending ops as committed as we can.
    //
    // We may have some pending ops that, according to the leader, are now
    // committed. We Apply them early, because:
    // - Soon (step 2) we may reject the call due to excessive memory pressure. One
    //   way to relieve the pressure is by flushing the MRS, and applying these
    //   ops may unblock an in-flight Flush().
    // - The Apply and subsequent Prepares (step 2) can take place concurrently.
    //
    // 2 - We enqueue the Prepare of the ops.
    //
    // The actual prepares are enqueued in order but happen asynchronously so we don't
    // have decoding/acquiring locks on the critical path.
    //
    // We need to do this now for a number of reasons:
    // - Prepares, by themselves, are inconsequential, i.e. they do not mutate the
    //   state machine so, were we to crash afterwards, having the prepares in-flight
    //   won't hurt.
    // - Prepares depend on factors external to consensus (the op drivers and
    //   the TabletReplica) so if for some reason they cannot be enqueued we must know
    //   before we try write them to the WAL. Once enqueued, we assume that prepare will
    //   always succeed on a replica op (because the leader already prepared them
    //   successfully, and thus we know they are valid).
    // - The prepares corresponding to every operation that was logged must be in-flight
    //   first. This because should we need to abort certain ops (say a new leader
    //   says they are not committed) we need to have those prepares in-flight so that
    //   the ops can be continued (in the abort path).
    // - Failure to enqueue prepares is OK, we can continue and let the leader know that
    //   we only went so far. The leader will re-send the remaining messages.
    // - Prepares represent new ops, and ops consume memory. Thus, if the
    //   overall memory pressure on the server is too high, we will reject the prepares.
    //
    // 3 - We enqueue the writes to the WAL.
    //
    // We enqueue writes to the WAL, but only the operations that were successfully
    // enqueued for prepare (for the reasons introduced above). This means that even
    // if a prepare fails to enqueue, if any of the previous prepares were successfully
    // submitted they must be written to the WAL.
    // If writing to the WAL fails, we're in an inconsistent state and we crash. In this
    // case, no one will ever know of the ops we previously prepared so those are
    // inconsequential.
    //
    // 4 - We mark the ops as committed.
    //
    // For each op which has been committed by the leader, we update the
    // op state to reflect that. If the logging has already succeeded for that
    // op, this will trigger the Apply phase. Otherwise, Apply will be triggered
    // when the logging completes. In both cases the Apply phase executes asynchronously.
    // This must, of course, happen after the prepares have been triggered as the same batch
    // can both replicate/prepare and commit/apply an operation.
    //
    // Currently, if a prepare failed to enqueue we still trigger all applies for operations
    // with an id lower than it (if we have them). This is important now as the leader will
    // not re-send those commit messages. This will be moot when we move to the commit
    // commitIndex way of doing things as we can simply ignore the applies as we know
    // they will be triggered with the next successful batch.
    //
    // 5 - We wait for the writes to be durable.
    //
    // Before replying to the leader we wait for the writes to be durable. We then
    // just update the last replicated watermark and respond.
    //
    // TODO - These failure scenarios need to be exercised in an unit
    //        test. Moreover we need to add more fault injection spots (well that
    //        and actually use the) for each of these steps.
    //        This will be done in a follow up patch.
    TRACE("Updating replica for $0 ops", request->ops_size());
  
    // The deduplicated request.
    LeaderRequest deduped_req;
    auto& messages = deduped_req.messages;
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      RETURN_NOT_OK(CheckRunningUnlocked());
      if (!cmeta_->IsMemberInConfig(peer_uuid(), ACTIVE_CONFIG)) {
        LOG_WITH_PREFIX_UNLOCKED(INFO) << "Allowing update even though not a member of the config";
      }
  
      deduped_req.leader_uuid = request->caller_uuid();
  
      RETURN_NOT_OK(CheckLeaderRequestUnlocked(request, response, &deduped_req));
      if (response->status().has_error()) {
        // We had an error, like an invalid term, we still fill the response.
        FillConsensusResponseOKUnlocked(response);
        return Status::OK();
      }
  
      // As soon as we decide to accept the message:
      //   * snooze the failure detector
      //   * prohibit voting for anyone for the minimum election timeout
      // We are guaranteed to be acting as a FOLLOWER at this point by the above
      // sanity check.
      SnoozeFailureDetector();
      WithholdVotes();
  
      last_leader_communication_time_micros_ = GetMonoTimeMicros();
  
      // Reset the 'failed_elections_since_stable_leader' metric now that we've
      // accepted an update from the established leader. This is done in addition
      // to the reset of the value in SetLeaderUuidUnlocked() because there is
      // a potential race between resetting the failed elections count in
      // SetLeaderUuidUnlocked() and incrementing after a failed election
      // if another replica was elected leader in an election concurrent with
      // the one called by this replica.
      failed_elections_since_stable_leader_ = 0;
      num_failed_elections_metric_->set_value(failed_elections_since_stable_leader_);
  
      // We update the lag metrics here in addition to after appending to the queue so the
      // metrics get updated even when the operation is rejected.
      queue_->UpdateLastIndexAppendedToLeader(request->last_idx_appended_to_leader());
  
      // 1 - Early commit pending (and committed) ops
  
      // What should we commit?
      // 1. As many pending ops as we can, except...
      // 2. ...if we commit beyond the preceding index, we'd regress KUDU-639, and...
      // 3. ...the leader's committed index is always our upper bound.
      const int64_t early_apply_up_to = std::min({
          pending_->GetLastPendingOpOpId().index(),
          deduped_req.preceding_opid->index(),
          request->committed_index()});
  
      VLOG_WITH_PREFIX_UNLOCKED(1) << "Early marking committed up to " << early_apply_up_to
                                   << ", Last pending opid index: "
                                   << pending_->GetLastPendingOpOpId().index()
                                   << ", preceding opid index: "
                                   << deduped_req.preceding_opid->index()
                                   << ", requested index: " << request->committed_index();
      TRACE("Early marking committed up to index $0", early_apply_up_to);
      CHECK_OK(pending_->AdvanceCommittedIndex(early_apply_up_to));
  
      // 2 - Enqueue the prepares
  
      TRACE("Triggering prepare for $0 ops", messages.size());
  
      if (PREDICT_TRUE(!messages.empty())) {
        // This request contains at least one message, and is likely to increase
        // our memory pressure.
        double capacity_pct;
        if (process_memory::SoftLimitExceeded(&capacity_pct)) {
          if (follower_memory_pressure_rejections_) follower_memory_pressure_rejections_->Increment();
          string msg = StringPrintf(
              "Soft memory limit exceeded (at %.2f%% of capacity)",
              capacity_pct);
          if (capacity_pct >= FLAGS_memory_limit_warn_threshold_percentage) {
            KLOG_EVERY_N_SECS(WARNING, 1) << "Rejecting consensus request: " << msg
                                          << THROTTLE_MSG;
          } else {
            KLOG_EVERY_N_SECS(INFO, 1) << "Rejecting consensus request: " << msg
                                       << THROTTLE_MSG;
          }
          return Status::ServiceUnavailable(msg);
        }
      }
  
      Status prepare_status;
      auto iter = messages.begin();
      while (iter != messages.end()) {
        prepare_status = StartFollowerOpUnlocked(*iter);
        if (PREDICT_FALSE(!prepare_status.ok())) {
          break;
        }
        // TODO(dralves) Without leader leases this shouldn't be allowed to fail.
        // Once we have that functionality we'll have to revisit this.
        CHECK_OK(time_manager_->MessageReceivedFromLeader(*(*iter)->get()));
        ++iter;
      }
  
      // If we stopped before reaching the end we failed to prepare some message(s) and need
      // to perform cleanup, namely trimming deduped_req.messages to only contain the messages
      // that were actually prepared, and deleting the other ones since we've taken ownership
      // when we first deduped.
      if (iter != messages.end()) {
        LOG_WITH_PREFIX_UNLOCKED(WARNING) << Substitute(
            "Could not prepare op '$0' and following $1 ops. "
            "Status for this op: $2",
            (*iter)->get()->id().ShortDebugString(),
            std::distance(iter, messages.end()) - 1,
            prepare_status.ToString());
        iter = messages.erase(iter, messages.end());
  
        // If this is empty, it means we couldn't prepare a single de-duped message. There is nothing
        // else we can do. The leader will detect this and retry later.
        if (messages.empty()) {
          string msg = Substitute("Rejecting Update request from peer $0 for term $1. "
                                  "Could not prepare a single op due to: $2",
                                  request->caller_uuid(),
                                  request->caller_term(),
                                  prepare_status.ToString());
          LOG_WITH_PREFIX_UNLOCKED(INFO) << msg;
          FillConsensusResponseError(response, ConsensusErrorPB::CANNOT_PREPARE,
                                     Status::IllegalState(msg));
          FillConsensusResponseOKUnlocked(response);
          return Status::OK();
        }
      }
  
      // All ops that are going to be prepared were started, advance the safe timestamp.
      // TODO(dralves) This is only correct because the queue only sets safe time when the request is
      // an empty heartbeat. If we actually start setting this on a consensus request along with
      // actual messages we need to be careful to ignore it if any of the messages fails to prepare.
      if (request->has_safe_timestamp()) {
        time_manager_->AdvanceSafeTime(Timestamp(request->safe_timestamp()));
      }
  
      OpId last_from_leader;
      // 3 - Enqueue the writes.
      // Now that we've triggered the prepares enqueue the operations to be written
      // to the WAL.
      if (PREDICT_TRUE(!messages.empty())) {
        last_from_leader = messages.back()->get()->id();
        // Trigger the log append asap, if fsync() is on this might take a while
        // and we can't reply until this is done.
        //
        // Since we've prepared, we need to be able to append (or we risk trying to apply
        // later something that wasn't logged). We crash if we can't.
        CHECK_OK(queue_->AppendOperations(messages, sync_status_cb));
      } else {
        last_from_leader = *deduped_req.preceding_opid;
      }
  
      // 4 - Mark ops as committed
  
      // Choose the last operation to be applied. This will either be 'committed_index', if
      // no prepare enqueuing failed, or the minimum between 'committed_index' and the id of
      // the last successfully enqueued prepare, if some prepare failed to enqueue.
      int64_t apply_up_to;
      if (last_from_leader.index() < request->committed_index()) {
        // we should never apply anything later than what we received in this request
        apply_up_to = last_from_leader.index();
  
        VLOG_WITH_PREFIX_UNLOCKED(2) << "Received commit index "
            << request->committed_index() << " from the leader but only"
            << " marked up to " << apply_up_to << " as committed.";
      } else {
        apply_up_to = request->committed_index();
      }
  
      VLOG_WITH_PREFIX_UNLOCKED(1) << "Marking committed up to " << apply_up_to;
      TRACE("Marking committed up to $0", apply_up_to);
      CHECK_OK(pending_->AdvanceCommittedIndex(apply_up_to));
      queue_->UpdateFollowerWatermarks(apply_up_to, request->all_replicated_index());
  
      // If any messages failed to be started locally, then we already have removed them
      // from 'deduped_req' at this point. So, 'last_from_leader' is the last one that
      // we might apply.
      last_received_cur_leader_ = last_from_leader;
  
      // Fill the response with the current state. We will not mutate anymore state until
      // we actually reply to the leader, we'll just wait for the messages to be durable.
      FillConsensusResponseOKUnlocked(response);
    }
    // Release the lock while we wait for the log append to finish so that commits can go through.
    // We'll re-acquire it before we update the state again.
  
    // Update the last replicated op id
    if (!messages.empty()) {
  
      // 5 - We wait for the writes to be durable.
  
      // Note that this is safe because dist consensus now only supports a single outstanding
      // request at a time and this way we can allow commits to proceed while we wait.
      TRACE("Waiting on the replicates to finish logging");
      TRACE_EVENT0("consensus", "Wait for log");
      Status s;
      do {
        s = log_synchronizer.WaitFor(
            MonoDelta::FromMilliseconds(FLAGS_raft_heartbeat_interval_ms));
        // If just waiting for our log append to finish lets snooze the timer.
        // We don't want to fire leader election nor accept vote requests because
        // we're still processing the Raft message from the leader,
        // waiting on our own log.
        if (s.IsTimedOut()) {
          SnoozeFailureDetector();
          WithholdVotes();
        }
      } while (s.IsTimedOut());
      RETURN_NOT_OK(s);
  
      TRACE("finished");
    }
  
    VLOG_WITH_PREFIX(2) << "Replica updated. " << ToString()
                        << ". Request: " << SecureShortDebugString(*request);
  
    TRACE("UpdateReplicas() finished");
    return Status::OK();
  }
  
  void RaftConsensus::FillConsensusResponseOKUnlocked(ConsensusResponsePB* response) {
    DCHECK(lock_.is_locked());
    TRACE("Filling consensus response to leader.");
    response->set_responder_term(CurrentTermUnlocked());
    response->mutable_status()->mutable_last_received()->CopyFrom(
        queue_->GetLastOpIdInLog());
    response->mutable_status()->mutable_last_received_current_leader()->CopyFrom(
        last_received_cur_leader_);
    response->mutable_status()->set_last_committed_idx(
        queue_->GetCommittedIndex());
    if (PREDICT_TRUE(server_ctx_.quiescing) && server_ctx_.quiescing->load()) {
      response->set_server_quiescing(true);
    }
  }
  
  void RaftConsensus::FillConsensusResponseError(ConsensusResponsePB* response,
                                                 ConsensusErrorPB::Code error_code,
                                                 const Status& status) {
    ConsensusErrorPB* error = response->mutable_status()->mutable_error();
    error->set_code(error_code);
    StatusToPB(status, error->mutable_status());
  }
  
  Status RaftConsensus::RequestVote(const VoteRequestPB* request,
                                    TabletVotingState tablet_voting_state,
                                    VoteResponsePB* response) {
    TRACE_EVENT2("consensus", "RaftConsensus::RequestVote",
                 "peer", peer_uuid(),
                 "tablet", options_.tablet_id);
    response->set_responder_uuid(peer_uuid());
  
    // If we've heard recently from the leader, then we should ignore the request.
    // It might be from a "disruptive" server. This could happen in a few cases:
    //
    // 1) Network partitions
    // If the leader can talk to a majority of the nodes, but is partitioned from a
    // bad node, the bad node's failure detector will trigger. If the bad node is
    // able to reach other nodes in the cluster, it will continuously trigger elections.
    //
    // 2) An abandoned node
    // It's possible that a node has fallen behind the log GC mark of the leader. In that
    // case, the leader will stop sending it requests. Eventually, the the configuration
    // will change to eject the abandoned node, but until that point, we don't want the
    // abandoned follower to disturb the other nodes.
    //
    // 3) Other dynamic scenarios with a stale former leader
    // This is a generalization of the case 1. It's possible that a stale former
    // leader detects it's not a leader anymore at some point, but a majority
    // of replicas has elected a new leader already.
    //
    // See also https://ramcloud.stanford.edu/~ongaro/thesis.pdf
    // section 4.2.3.
    if (PREDICT_TRUE(!request->ignore_live_leader()) &&
        MonoTime::Now() < withhold_votes_until_) {
      return RequestVoteRespondLeaderIsAlive(request, response);
    }
  
    // We must acquire the update lock in order to ensure that this vote action
    // takes place between requests.
    // Lock ordering: update_lock_ must be acquired before lock_.
    std::unique_lock<simple_spinlock> update_guard(update_lock_, std::defer_lock);
    if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
      update_guard.try_lock();
    } else {
      // If failure detection is not enabled, then we can't just reject the vote,
      // because there will be no automatic retry later. So, block for the lock.
      update_guard.lock();
    }
    if (!update_guard.owns_lock()) {
      // There is another vote or update concurrent with the vote. In that case, that
      // other request is likely to reset the timer, and we'll end up just voting
      // "NO" after waiting. To avoid starving RPC handlers and causing cascading
      // timeouts, just vote a quick NO.
      return RequestVoteRespondIsBusy(request, response);
    }
  
    // Acquire the replica state lock so we can read / modify the consensus state.
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
  
    // Ensure our lifecycle state is compatible with voting.
    // If RaftConsensus is running, we use the latest OpId from the WAL to vote.
    // Otherwise, we must be voting while tombstoned.
    OpId local_last_logged_opid;
    switch (state_) {
      case kShutdown:
        return Status::IllegalState("cannot vote while shut down");
      case kRunning:
        // Note: it is (theoretically) possible for 'tombstone_last_logged_opid'
        // to be passed in and by the time we reach here the state is kRunning.
        // That may occur when a vote request comes in at the end of a tablet
        // copy and then tablet bootstrap completes quickly. In that case, we
        // ignore the passed-in value and use the latest OpId from our queue.
        local_last_logged_opid = queue_->GetLastOpIdInLog();
        break;
      default:
        if (!tablet_voting_state.tombstone_last_logged_opid_) {
          return Status::IllegalState("must be running to vote when last-logged opid is not known");
        }
        if (!FLAGS_raft_enable_tombstoned_voting) {
          return Status::IllegalState("must be running to vote when tombstoned voting is disabled");
        }
        local_last_logged_opid = *(tablet_voting_state.tombstone_last_logged_opid_);
        if (tablet_voting_state.data_state_ == tablet::TABLET_DATA_COPYING) {
          LOG_WITH_PREFIX_UNLOCKED(INFO) << "voting while copying based on last-logged opid "
                                         << local_last_logged_opid;
        } else if (tablet_voting_state.data_state_ == tablet::TABLET_DATA_TOMBSTONED) {
          LOG_WITH_PREFIX_UNLOCKED(INFO) << "voting while tombstoned based on last-logged opid "
                                         << local_last_logged_opid;
        }
        break;
    }
    DCHECK(local_last_logged_opid.IsInitialized());
  
    // If the node is not in the configuration, allow the vote (this is required by Raft)
    // but log an informational message anyway.
    if (!cmeta_->IsMemberInConfig(request->candidate_uuid(), ACTIVE_CONFIG)) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Handling vote request from an unknown peer "
                                     << request->candidate_uuid();
    }
  
    // Recheck the heard-from-the-leader condition. There is a slight chance
    // that a heartbeat from the leader replica registers after the first check
    // in the very beginning of this method and before lock_ is acquired. This
    // extra check costs a little, but it helps in avoiding extra election rounds
    // and disruptive transfers of the replica leadership.
    if (PREDICT_TRUE(!request->ignore_live_leader()) &&
        MonoTime::Now() < withhold_votes_until_) {
      return RequestVoteRespondLeaderIsAlive(request, response);
    }
  
    // Candidate is running behind.
    if (request->candidate_term() < CurrentTermUnlocked()) {
      return RequestVoteRespondInvalidTerm(request, response);
    }
  
    // We already voted this term.
    if (request->candidate_term() == CurrentTermUnlocked() &&
        HasVotedCurrentTermUnlocked()) {
  
      // Already voted for the same candidate in the current term.
      if (GetVotedForCurrentTermUnlocked() == request->candidate_uuid()) {
        return RequestVoteRespondVoteAlreadyGranted(request, response);
      }
  
      // Voted for someone else in current term.
      return RequestVoteRespondAlreadyVotedForOther(request, response);
    }
  
    // Candidate must have last-logged OpId at least as large as our own to get
    // our vote.
    bool vote_yes = !OpIdLessThan(request->candidate_status().last_received(),
                                  local_last_logged_opid);
  
    // Record the term advancement if necessary. We don't do so in the case of
    // pre-elections because it's possible that the node who called the pre-election
    // has actually now successfully become leader of the prior term, in which case
    // bumping our term here would disrupt it.
    if (!request->is_pre_election() &&
        request->candidate_term() > CurrentTermUnlocked()) {
      // If we are going to vote for this peer, then we will flush the consensus metadata
      // to disk below when we record the vote, and we can skip flushing the term advancement
      // to disk here.
      auto flush = vote_yes ? SKIP_FLUSH_TO_DISK : FLUSH_TO_DISK;
      RETURN_NOT_OK_PREPEND(HandleTermAdvanceUnlocked(request->candidate_term(), flush),
          Substitute("Could not step down in RequestVote. Current term: $0, candidate term: $1",
                     CurrentTermUnlocked(), request->candidate_term()));
    }
  
    if (!vote_yes) {
      return RequestVoteRespondLastOpIdTooOld(local_last_logged_opid, request, response);
    }
  
    // Passed all our checks. Vote granted.
    return RequestVoteRespondVoteGranted(request, response);
  }
  
  Status RaftConsensus::ChangeConfig(const ChangeConfigRequestPB& req,
                                     StatusCallback client_cb,
                                     boost::optional<TabletServerErrorPB::Code>* error_code) {
    TRACE_EVENT2("consensus", "RaftConsensus::ChangeConfig",
                 "peer", peer_uuid(),
                 "tablet", options_.tablet_id);
  
    BulkChangeConfigRequestPB bulk_req;
    *bulk_req.mutable_tablet_id() = req.tablet_id();
  
    if (req.has_dest_uuid()) {
      *bulk_req.mutable_dest_uuid() = req.dest_uuid();
    }
    if (req.has_cas_config_opid_index()) {
      bulk_req.set_cas_config_opid_index(req.cas_config_opid_index());
    }
    auto* change = bulk_req.add_config_changes();
    if (req.has_type()) {
      change->set_type(req.type());
    }
    if (req.has_server()) {
      *change->mutable_peer() = req.server();
    }
  
    return BulkChangeConfig(bulk_req, std::move(client_cb), error_code);
  }
  
  Status RaftConsensus::BulkChangeConfig(const BulkChangeConfigRequestPB& req,
                                         StatusCallback client_cb,
                                         boost::optional<TabletServerErrorPB::Code>* error_code) {
    TRACE_EVENT2("consensus", "RaftConsensus::BulkChangeConfig",
                 "peer", peer_uuid(),
                 "tablet", options_.tablet_id);
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      RETURN_NOT_OK(CheckRunningUnlocked());
      RETURN_NOT_OK(CheckActiveLeaderUnlocked());
      RETURN_NOT_OK(CheckNoConfigChangePendingUnlocked());
  
      // We are required by Raft to reject config change operations until we have
      // committed at least one operation in our current term as leader.
      // See https://groups.google.com/forum/#!topic/raft-dev/t4xj6dJTP6E
      if (!queue_->IsCommittedIndexInCurrentTerm()) {
        return Status::IllegalState("Leader has not yet committed an operation in its own term");
      }
  
      const RaftConfigPB committed_config = cmeta_->CommittedConfig();
  
      // Support atomic ChangeConfig requests.
      if (req.has_cas_config_opid_index()) {
        if (committed_config.opid_index() != req.cas_config_opid_index()) {
          *error_code = TabletServerErrorPB::CAS_FAILED;
          return Status::IllegalState(Substitute("Request specified cas_config_opid_index "
                                                 "of $0 but the committed config has opid_index "
                                                 "of $1",
                                                 req.cas_config_opid_index(),
                                                 committed_config.opid_index()));
        }
      }
  
      // 'new_config' will be modified in-place and validated before being used
      // as the new Raft configuration.
      RaftConfigPB new_config = committed_config;
  
      // Enforce the "one by one" config change rules, even with the bulk API.
      // Keep track of total voters added, including non-voters promoted to
      // voters, and removed, including voters demoted to non-voters.
      int num_voters_modified = 0;
  
      // A record of the peers being modified so that we can enforce only one
      // change per peer per request.
      unordered_set<string> peers_modified;
  
      for (const auto& item : req.config_changes()) {
        if (PREDICT_FALSE(!item.has_type())) {
          *error_code = TabletServerErrorPB::INVALID_CONFIG;
          return Status::InvalidArgument("Must specify 'type' argument",
                                         SecureShortDebugString(req));
        }
        if (PREDICT_FALSE(!item.has_peer())) {
          *error_code = TabletServerErrorPB::INVALID_CONFIG;
          return Status::InvalidArgument("Must specify 'peer' argument",
                                         SecureShortDebugString(req));
        }
  
        ChangeConfigType type = item.type();
        const RaftPeerPB& peer = item.peer();
  
        if (PREDICT_FALSE(!peer.has_permanent_uuid())) {
          return Status::InvalidArgument("peer must have permanent_uuid specified",
                                         SecureShortDebugString(req));
        }
  
        if (!InsertIfNotPresent(&peers_modified, peer.permanent_uuid())) {
          return Status::InvalidArgument(
              Substitute("only one change allowed per peer: peer $0 appears more "
                         "than once in the config change request",
                         peer.permanent_uuid()),
              SecureShortDebugString(req));
        }
  
        const string& server_uuid = peer.permanent_uuid();
        switch (type) {
          case ADD_PEER:
            // Ensure the peer we are adding is not already a member of the configuration.
            if (IsRaftConfigMember(server_uuid, committed_config)) {
              return Status::InvalidArgument(
                  Substitute("Server with UUID $0 is already a member of the config. RaftConfig: $1",
                             server_uuid, SecureShortDebugString(committed_config)));
            }
            if (!peer.has_member_type()) {
              return Status::InvalidArgument("peer must have member_type specified",
                                             SecureShortDebugString(req));
            }
            if (!peer.has_last_known_addr()) {
              return Status::InvalidArgument("peer must have last_known_addr specified",
                                             SecureShortDebugString(req));
            }
            if (peer.member_type() == RaftPeerPB::VOTER) {
              num_voters_modified++;
            }
            *new_config.add_peers() = peer;
            break;
  
          case REMOVE_PEER:
            if (server_uuid == peer_uuid()) {
              return Status::InvalidArgument(
                  Substitute("Cannot remove peer $0 from the config because it is the leader. "
                             "Force another leader to be elected to remove this peer. "
                             "Consensus state: $1",
                             server_uuid,
                             SecureShortDebugString(cmeta_->ToConsensusStatePB())));
            }
            if (!RemoveFromRaftConfig(&new_config, server_uuid)) {
              return Status::NotFound(
                  Substitute("Server with UUID $0 not a member of the config. RaftConfig: $1",
                             server_uuid, SecureShortDebugString(committed_config)));
            }
            if (IsRaftConfigVoter(server_uuid, committed_config)) {
              num_voters_modified++;
            }
            break;
  
          case MODIFY_PEER: {
            RaftPeerPB* modified_peer;
            RETURN_NOT_OK(GetRaftConfigMember(&new_config, server_uuid, &modified_peer));
            const RaftPeerPB orig_peer(*modified_peer);
            // Override 'member_type' and items within 'attrs' only if they are
            // explicitly passed in the request. At least one field must be
            // modified to be a valid request.
            if (peer.has_member_type() && peer.member_type() != modified_peer->member_type()) {
              if (modified_peer->member_type() == RaftPeerPB::VOTER ||
                  peer.member_type() == RaftPeerPB::VOTER) {
                // This is a 'member_type' change involving a VOTER, i.e. a
                // promotion or demotion.
                num_voters_modified++;
              }
              // A leader must be forced to step down before demoting it.
              if (server_uuid == peer_uuid()) {
                return Status::InvalidArgument(
                    Substitute("Cannot modify member type of peer $0 because it is the leader. "
                                "Cause another leader to be elected to modify this peer. "
                                "Consensus state: $1",
                                server_uuid,
                                SecureShortDebugString(cmeta_->ToConsensusStatePB())));
              }
              modified_peer->set_member_type(peer.member_type());
            }
            if (peer.attrs().has_promote()) {
              modified_peer->mutable_attrs()->set_promote(peer.attrs().promote());
            }
            if (peer.attrs().has_replace()) {
              modified_peer->mutable_attrs()->set_replace(peer.attrs().replace());
            }
            // Ensure that MODIFY_PEER actually modified something.
            if (MessageDifferencer::Equals(orig_peer, *modified_peer)) {
              return Status::InvalidArgument("must modify a field when calling MODIFY_PEER");
            }
            break;
          }
  
          default:
            return Status::NotSupported(Substitute(
                "$0: unsupported type of configuration change",
                ChangeConfigType_Name(type)));
        }
      }
  
      // Don't allow no-op config changes to be committed.
      if (MessageDifferencer::Equals(committed_config, new_config)) {
        return Status::InvalidArgument("requested configuration change does not "
                                       "actually modify the config",
                                       SecureShortDebugString(req));
      }
  
      // Ensure this wasn't an illegal bulk change.
      if (num_voters_modified > 1) {
        return Status::InvalidArgument("it is not safe to modify the VOTER status "
                                       "of more than one peer at a time",
                                       SecureShortDebugString(req));
      }
  
      // We'll assign a new opid_index to this config change.
      new_config.clear_opid_index();
  
      RETURN_NOT_OK(ReplicateConfigChangeUnlocked(
          committed_config, std::move(new_config),
          [this, client_cb](const Status& s) {
            this->MarkDirtyOnSuccess("Config change replication complete",
                                     client_cb, s);
          }));
    } // Release lock before signaling request.
    peer_manager_->SignalRequest();
    return Status::OK();
  }
  
  Status RaftConsensus::UnsafeChangeConfig(
      const UnsafeChangeConfigRequestPB& req,
      boost::optional<tserver::TabletServerErrorPB::Code>* error_code) {
    if (PREDICT_FALSE(!req.has_new_config())) {
      *error_code = TabletServerErrorPB::INVALID_CONFIG;
      return Status::InvalidArgument("Request must contain 'new_config' argument "
                                     "to UnsafeChangeConfig()", SecureShortDebugString(req));
    }
    if (PREDICT_FALSE(!req.has_caller_id())) {
      *error_code = TabletServerErrorPB::INVALID_CONFIG;
      return Status::InvalidArgument("Must specify 'caller_id' argument to UnsafeChangeConfig()",
                                     SecureShortDebugString(req));
    }
  
    // Grab the committed config and current term on this node.
    int64_t current_term;
    RaftConfigPB committed_config;
    int64_t all_replicated_index;
    int64_t last_committed_index;
    OpId preceding_opid;
    uint64_t msg_timestamp;
    {
      // Take the snapshot of the replica state and queue state so that
      // we can stick them in the consensus update request later.
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      current_term = CurrentTermUnlocked();
      committed_config = cmeta_->CommittedConfig();
      if (cmeta_->has_pending_config()) {
        LOG_WITH_PREFIX_UNLOCKED(WARNING)
              << "Replica has a pending config, but the new config "
              << "will be unsafely changed anyway. "
              << "Currently pending config on the node: "
              << SecureShortDebugString(cmeta_->PendingConfig());
      }
      all_replicated_index = queue_->GetAllReplicatedIndex();
      last_committed_index = queue_->GetCommittedIndex();
      preceding_opid = queue_->GetLastOpIdInLog();
      msg_timestamp = time_manager_->GetSerialTimestamp().value();
    }
  
    // Validate that passed replica uuids are part of the committed config
    // on this node.  This allows a manual recovery tool to only have to specify
    // the uuid of each replica in the new config without having to know the
    // addresses of each server (since we can get the address information from
    // the committed config). Additionally, only a subset of the committed config
    // is required for typical cluster repair scenarios.
    std::unordered_set<string> retained_peer_uuids;
    const RaftConfigPB& config = req.new_config();
    for (const RaftPeerPB& new_peer : config.peers()) {
      const string& peer_uuid = new_peer.permanent_uuid();
      retained_peer_uuids.insert(peer_uuid);
      if (!IsRaftConfigMember(peer_uuid, committed_config)) {
        *error_code = TabletServerErrorPB::INVALID_CONFIG;
        return Status::InvalidArgument(Substitute("Peer with uuid $0 is not in the committed  "
                                                  "config on this replica, rejecting the  "
                                                  "unsafe config change request for tablet $1. "
                                                  "Committed config: $2",
                                                  peer_uuid, req.tablet_id(),
                                                  SecureShortDebugString(committed_config)));
      }
    }
  
    RaftConfigPB new_config = committed_config;
    for (const auto& peer : committed_config.peers()) {
      const string& peer_uuid = peer.permanent_uuid();
      if (!ContainsKey(retained_peer_uuids, peer_uuid)) {
        CHECK(RemoveFromRaftConfig(&new_config, peer_uuid));
      }
    }
    // Check that local peer is part of the new config and is a VOTER.
    // Although it is valid for a local replica to not have itself
    // in the committed config, it is rare and a replica without itself
    // in the latest config is definitely not caught up with the latest leader's log.
    if (!IsRaftConfigVoter(peer_uuid(), new_config)) {
      *error_code = TabletServerErrorPB::INVALID_CONFIG;
      return Status::InvalidArgument(Substitute("Local replica uuid $0 is not "
                                                "a VOTER in the new config, "
                                                "rejecting the unsafe config "
                                                "change request for tablet $1. "
                                                "Rejected config: $2" ,
                                                peer_uuid(), req.tablet_id(),
                                                SecureShortDebugString(new_config)));
    }
    new_config.set_unsafe_config_change(true);
    int64_t replicate_opid_index = preceding_opid.index() + 1;
    new_config.set_opid_index(replicate_opid_index);
  
    // Sanity check the new config. 'type' is irrelevant here.
    Status s = VerifyRaftConfig(new_config);
    if (!s.ok()) {
      *error_code = TabletServerErrorPB::INVALID_CONFIG;
      return Status::InvalidArgument(Substitute("The resulting new config for tablet $0  "
                                                "from passed parameters has failed raft "
                                                "config sanity check: $1",
                                                req.tablet_id(), s.ToString()));
    }
  
    // Prepare the consensus request as if the request is being generated
    // from a different leader.
    ConsensusRequestPB consensus_req;
    consensus_req.set_caller_uuid(req.caller_id());
    // Bumping up the term for the consensus request being generated.
    // This makes this request appear to come from a new leader that
    // the local replica doesn't know about yet. If the local replica
    // happens to be the leader, this will cause it to step down.
    const int64_t new_term = current_term + 1;
    consensus_req.set_caller_term(new_term);
    consensus_req.mutable_preceding_id()->CopyFrom(preceding_opid);
    consensus_req.set_committed_index(last_committed_index);
    consensus_req.set_all_replicated_index(all_replicated_index);
  
    // Prepare the replicate msg to be replicated.
    ReplicateMsg* replicate = consensus_req.add_ops();
    ChangeConfigRecordPB* cc_req = replicate->mutable_change_config_record();
    cc_req->set_tablet_id(req.tablet_id());
    *cc_req->mutable_old_config() = committed_config;
    *cc_req->mutable_new_config() = new_config;
    OpId* id = replicate->mutable_id();
    // Bumping up both the term and the opid_index from what's found in the log.
    id->set_term(new_term);
    id->set_index(replicate_opid_index);
    replicate->set_op_type(CHANGE_CONFIG_OP);
    replicate->set_timestamp(msg_timestamp);
  
    VLOG_WITH_PREFIX(3) << "UnsafeChangeConfig: Generated consensus request: "
                        << SecureShortDebugString(consensus_req);
  
    LOG_WITH_PREFIX(WARNING)
          << "PROCEEDING WITH UNSAFE CONFIG CHANGE ON THIS SERVER, "
          << "COMMITTED CONFIG: " << SecureShortDebugString(committed_config)
          << "NEW CONFIG: " << SecureShortDebugString(new_config);
  
    ConsensusResponsePB consensus_resp;
    return Update(&consensus_req, &consensus_resp).AndThen([&consensus_resp]{
      return consensus_resp.has_error()
          ? StatusFromPB(consensus_resp.error().status()) : Status::OK();
    });
  }
  
  void RaftConsensus::Stop() {
    TRACE_EVENT2("consensus", "RaftConsensus::Shutdown",
                 "peer", peer_uuid(),
                 "tablet", options_.tablet_id);
  
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      if (state_ == kStopping || state_ == kStopped || state_ == kShutdown) return;
      // Transition to kStopping state.
      SetStateUnlocked(kStopping);
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Raft consensus shutting down.";
    }
  
    // Close the peer manager.
    if (peer_manager_) peer_manager_->Close();
  
    // We must close the queue after we close the peers.
    if (queue_) {
      // If we were leader, decrement the number of leaders there are now.
      if (queue_->IsInLeaderMode() && server_ctx_.num_leaders) {
        server_ctx_.num_leaders->IncrementBy(-1);
      }
      queue_->Close();
    }
  
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      if (pending_) CHECK_OK(pending_->CancelPendingOps());
      SetStateUnlocked(kStopped);
  
      // Clear leader status on Stop(), in case this replica was the leader. If
      // we don't do this, the log messages still show this node as the leader.
      // No need to sync it since it's not persistent state.
      if (cmeta_) {
        ClearLeaderUnlocked();
      }
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Raft consensus is shut down!";
    }
  
    // If we were the leader, stop withholding votes.
    auto max_time = MonoTime::Max();
    withhold_votes_until_.compare_exchange_strong(max_time, MonoTime::Min());
  
  
    // Shut down things that might acquire locks during destruction.
    if (raft_pool_token_) raft_pool_token_->Shutdown();
    if (failure_detector_) DisableFailureDetector();
  }
  
  void RaftConsensus::Shutdown() {
    // Avoid taking locks if already shut down so we don't violate
    // ThreadRestrictions assertions in the case where the RaftConsensus
    // destructor runs on the reactor thread due to an election callback being
    // the last outstanding reference.
    if (shutdown_.Load(kMemOrderAcquire)) return;
  
    Stop();
    {
      LockGuard l(lock_);
      SetStateUnlocked(kShutdown);
    }
    shutdown_.Store(true, kMemOrderRelease);
  }
  
  Status RaftConsensus::StartConsensusOnlyRoundUnlocked(const ReplicateRefPtr& msg) {
    DCHECK(lock_.is_locked());
    OperationType op_type = msg->get()->op_type();
    CHECK(IsConsensusOnlyOperation(op_type))
        << "Expected a consensus-only op type, got " << OperationType_Name(op_type)
        << ": " << SecureShortDebugString(*msg->get());
    VLOG_WITH_PREFIX_UNLOCKED(1) << "Starting consensus round: "
                                 << SecureShortDebugString(msg->get()->id());
    auto client_cb = [this](const Status& s) {
      this->MarkDirtyOnSuccess("Replicated consensus-only round",
                               &DoNothingStatusCB, s);
    };
    scoped_refptr<ConsensusRound> round(new ConsensusRound(this, msg));
    auto* round_raw = round.get();
    round->SetConsensusReplicatedCallback(
        [this, round_raw, client_cb](const Status& s) {
          this->NonTxRoundReplicationFinished(round_raw, client_cb, s);
        });
    return AddPendingOperationUnlocked(round);
  }
  
  Status RaftConsensus::AdvanceTermForTests(int64_t new_term) {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    CHECK_OK(CheckRunningUnlocked());
    return HandleTermAdvanceUnlocked(new_term);
  }
  
  string RaftConsensus::GetRequestVoteLogPrefixThreadSafe(const VoteRequestPB& request) const {
    return Substitute("$0Leader $1election vote request",
                      LogPrefixThreadSafe(),
                      request.is_pre_election() ? "pre-" : "");
  }
  
  string RaftConsensus::GetRequestVoteLogPrefixUnlocked(const VoteRequestPB& request) const {
    DCHECK(lock_.is_locked());
    return Substitute("$0Leader $1election vote request",
                      LogPrefixUnlocked(),
                      request.is_pre_election() ? "pre-" : "");
  }
  
  void RaftConsensus::FillVoteResponseVoteGranted(VoteResponsePB* response) {
    response->set_responder_term(CurrentTermUnlocked());
    response->set_vote_granted(true);
  }
  
  void RaftConsensus::FillVoteResponseVoteDenied(
      ConsensusErrorPB::Code error_code,
      VoteResponsePB* response,
      ResponderTermPolicy responder_term_policy) {
    response->set_vote_granted(false);
    response->mutable_consensus_error()->set_code(error_code);
    if (responder_term_policy == ResponderTermPolicy::SET) {
      response->set_responder_term(CurrentTermUnlocked());
    }
  }
  
  Status RaftConsensus::RequestVoteRespondInvalidTerm(const VoteRequestPB* request,
                                                      VoteResponsePB* response) {
    FillVoteResponseVoteDenied(ConsensusErrorPB::INVALID_TERM, response);
    string msg = Substitute("$0: Denying vote to candidate $1 for earlier term $2. "
                            "Current term is $3.",
                            GetRequestVoteLogPrefixUnlocked(*request),
                            request->candidate_uuid(),
                            request->candidate_term(),
                            CurrentTermUnlocked());
    LOG(INFO) << msg;
    StatusToPB(Status::InvalidArgument(msg), response->mutable_consensus_error()->mutable_status());
    return Status::OK();
  }
  
  Status RaftConsensus::RequestVoteRespondVoteAlreadyGranted(const VoteRequestPB* request,
                                                             VoteResponsePB* response) {
    FillVoteResponseVoteGranted(response);
    LOG(INFO) << Substitute("$0: Already granted yes vote for candidate $1 in term $2. "
                            "Re-sending same reply.",
                            GetRequestVoteLogPrefixUnlocked(*request),
                            request->candidate_uuid(),
                            request->candidate_term());
    return Status::OK();
  }
  
  Status RaftConsensus::RequestVoteRespondAlreadyVotedForOther(const VoteRequestPB* request,
                                                               VoteResponsePB* response) {
    FillVoteResponseVoteDenied(ConsensusErrorPB::ALREADY_VOTED, response);
    string msg = Substitute("$0: Denying vote to candidate $1 in current term $2: "
                            "Already voted for candidate $3 in this term.",
                            GetRequestVoteLogPrefixUnlocked(*request),
                            request->candidate_uuid(),
                            CurrentTermUnlocked(),
                            GetVotedForCurrentTermUnlocked());
    LOG(INFO) << msg;
    StatusToPB(Status::InvalidArgument(msg), response->mutable_consensus_error()->mutable_status());
    return Status::OK();
  }
  
  Status RaftConsensus::RequestVoteRespondLastOpIdTooOld(const OpId& local_last_logged_opid,
                                                         const VoteRequestPB* request,
                                                         VoteResponsePB* response) {
    FillVoteResponseVoteDenied(ConsensusErrorPB::LAST_OPID_TOO_OLD, response);
    string msg = Substitute("$0: Denying vote to candidate $1 for term $2 because "
                            "replica has last-logged OpId of $3, which is greater than that of the "
                            "candidate, which has last-logged OpId of $4.",
                            GetRequestVoteLogPrefixUnlocked(*request),
                            request->candidate_uuid(),
                            request->candidate_term(),
                            SecureShortDebugString(local_last_logged_opid),
                            SecureShortDebugString(request->candidate_status().last_received()));
    LOG(INFO) << msg;
    StatusToPB(Status::InvalidArgument(msg), response->mutable_consensus_error()->mutable_status());
    return Status::OK();
  }
  
  Status RaftConsensus::RequestVoteRespondLeaderIsAlive(const VoteRequestPB* request,
                                                        VoteResponsePB* response) {
    FillVoteResponseVoteDenied(ConsensusErrorPB::LEADER_IS_ALIVE, response,
                               ResponderTermPolicy::DO_NOT_SET);
    auto msg = Substitute("$0: Denying vote to candidate $1 for term $2 because "
                          "replica is either leader or believes a valid leader to "
                          "be alive.",
                          GetRequestVoteLogPrefixThreadSafe(*request),
                          request->candidate_uuid(),
                          request->candidate_term());
    VLOG(1) << msg;
    StatusToPB(Status::InvalidArgument(msg),
               response->mutable_consensus_error()->mutable_status());
    return Status::OK();
  }
  
  Status RaftConsensus::RequestVoteRespondIsBusy(
      const VoteRequestPB* request, VoteResponsePB* response) {
    // Don't set the term in the response: the requestor doesn't need it
    // to process the NO vote response in this case.
    FillVoteResponseVoteDenied(ConsensusErrorPB::CONSENSUS_BUSY, response,
                               ResponderTermPolicy::DO_NOT_SET);
    auto msg = Substitute("$0: Denying vote to candidate $1 for term $2 because "
                          "replica is already servicing an update from "
                          "a current leader or another vote",
                          GetRequestVoteLogPrefixThreadSafe(*request),
                          request->candidate_uuid(),
                          request->candidate_term());
    VLOG(1) << msg;
    StatusToPB(Status::ServiceUnavailable(msg),
               response->mutable_consensus_error()->mutable_status());
    return Status::OK();
  }
  
  Status RaftConsensus::RequestVoteRespondVoteGranted(const VoteRequestPB* request,
                                                      VoteResponsePB* response) {
    // We know our vote will be "yes", so avoid triggering an election while we
    // persist our vote to disk. We use an exponential backoff to avoid too much
    // split-vote contention when nodes display high latencies.
    MonoDelta backoff = LeaderElectionExpBackoffDeltaUnlocked();
    SnoozeFailureDetector(string("vote granted"), backoff);
  
    if (!request->is_pre_election()) {
      // Persist our vote to disk.
      RETURN_NOT_OK(SetVotedForCurrentTermUnlocked(request->candidate_uuid()));
    }
  
    FillVoteResponseVoteGranted(response);
  
    // Give peer time to become leader. Snooze one more time after persisting our
    // vote. When disk latency is high, this should help reduce churn.
    SnoozeFailureDetector(/*reason_for_log=*/boost::none, backoff);
  
    LOG(INFO) << Substitute("$0: Granting yes vote for candidate $1 in term $2.",
                            GetRequestVoteLogPrefixUnlocked(*request),
                            request->candidate_uuid(),
                            CurrentTermUnlocked());
    return Status::OK();
  }
  
  RaftPeerPB::Role RaftConsensus::role() const {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    return cmeta_->active_role();
  }
  
  int64_t RaftConsensus::CurrentTerm() const {
    LockGuard l(lock_);
    return CurrentTermUnlocked();
  }
  
  void RaftConsensus::SetStateUnlocked(State new_state) {
    switch (new_state) {
      case kInitialized:
        CHECK_EQ(kNew, state_);
        break;
      case kRunning:
        CHECK_EQ(kInitialized, state_);
        break;
      case kStopping:
        CHECK(state_ != kStopped && state_ != kShutdown) << "State = " << State_Name(state_);
        break;
      case kStopped:
        CHECK_EQ(kStopping, state_);
        break;
      case kShutdown:
        CHECK(state_ == kStopped || state_ == kShutdown) << "State = " << State_Name(state_);
        break;
      default:
        LOG(FATAL) << "Disallowed transition to state = " << State_Name(new_state);
        break;
    }
    state_ = new_state;
  }
  
  const char* RaftConsensus::State_Name(State state) {
    switch (state) {
      case kNew:
        return "New";
      case kInitialized:
        return "Initialized";
      case kRunning:
        return "Running";
      case kStopping:
        return "Stopping";
      case kStopped:
        return "Stopped";
      case kShutdown:
        return "Shut down";
      default:
        LOG(DFATAL) << "Unknown State value: " << state;
        return "Unknown";
    }
  }
  
  MonoDelta RaftConsensus::MinimumElectionTimeout() {
    int32_t failure_timeout = FLAGS_leader_failure_max_missed_heartbeat_periods *
        FLAGS_raft_heartbeat_interval_ms;
    return MonoDelta::FromMilliseconds(failure_timeout);
  }
  
  void RaftConsensus::SetLeaderUuidUnlocked(const string& uuid) {
    DCHECK(lock_.is_locked());
    failed_elections_since_stable_leader_ = 0;
    num_failed_elections_metric_->set_value(failed_elections_since_stable_leader_);
    cmeta_->set_leader_uuid(uuid);
    MarkDirty(Substitute("New leader $0", uuid));
  }
  
  Status RaftConsensus::ReplicateConfigChangeUnlocked(
      RaftConfigPB old_config,
      RaftConfigPB new_config,
      StatusCallback client_cb) {
    DCHECK(lock_.is_locked());
    auto cc_replicate = new ReplicateMsg();
    cc_replicate->set_op_type(CHANGE_CONFIG_OP);
    ChangeConfigRecordPB* cc_req = cc_replicate->mutable_change_config_record();
    cc_req->set_tablet_id(options_.tablet_id);
    *cc_req->mutable_old_config() = std::move(old_config);
    *cc_req->mutable_new_config() = std::move(new_config);
    CHECK_OK(time_manager_->AssignTimestamp(cc_replicate));
  
    scoped_refptr<ConsensusRound> round(
        new ConsensusRound(this, make_scoped_refptr(new RefCountedReplicate(cc_replicate))));
    auto* round_raw = round.get();
    round->SetConsensusReplicatedCallback(
        [this, round_raw, client_cb](const Status& s) {
          this->NonTxRoundReplicationFinished(round_raw, client_cb, s);
        });
    return AppendNewRoundToQueueUnlocked(round);
  }
  
  Status RaftConsensus::RefreshConsensusQueueAndPeersUnlocked() {
    DCHECK(lock_.is_locked());
    DCHECK_EQ(RaftPeerPB::LEADER, cmeta_->active_role());
    const RaftConfigPB& active_config = cmeta_->ActiveConfig();
  
    // Change the peers so that we're able to replicate messages remotely and
    // locally. The peer manager must be closed before updating the active config
    // in the queue -- when the queue is in LEADER mode, it checks that all
    // registered peers are a part of the active config.
    peer_manager_->Close();
    // TODO(todd): should use queue committed index here? in that case do
    // we need to pass it in at all?
    queue_->SetLeaderMode(pending_->GetCommittedIndex(),
                          CurrentTermUnlocked(),
                          active_config);
    RETURN_NOT_OK(peer_manager_->UpdateRaftConfig(active_config));
    return Status::OK();
  }
  
  const string& RaftConsensus::peer_uuid() const {
    return local_peer_pb_.permanent_uuid();
  }
  
  const string& RaftConsensus::tablet_id() const {
    return options_.tablet_id;
  }
  
  Status RaftConsensus::ConsensusState(ConsensusStatePB* cstate,
                                       IncludeHealthReport report_health) const {
    ThreadRestrictions::AssertWaitAllowed();
    UniqueLock l(lock_);
    if (state_ == kShutdown) {
      return Status::IllegalState("Tablet replica is shutdown");
    }
    ConsensusStatePB cstate_tmp = cmeta_->ToConsensusStatePB();
  
    // If we need to include the health report, merge it into the committed
    // config iff we believe we are the current leader of the config.
    if (report_health == INCLUDE_HEALTH_REPORT &&
        cmeta_->active_role() == RaftPeerPB::LEADER) {
      auto reports = queue_->ReportHealthOfPeers();
  
      // We don't need to access the queue anymore, so drop the consensus lock.
      l.unlock();
  
      // Iterate through each peer in the committed config and attach the health
      // report to it.
      RaftConfigPB* committed_raft_config = cstate_tmp.mutable_committed_config();
      for (int i = 0; i < committed_raft_config->peers_size(); i++) {
        RaftPeerPB* peer = committed_raft_config->mutable_peers(i);
        const HealthReportPB* report = FindOrNull(reports, peer->permanent_uuid());
        if (!report) continue; // Only attach details if we know about the peer.
        *peer->mutable_health_report() = *report;
      }
    }
    *cstate = std::move(cstate_tmp);
    return Status::OK();
  }
  
  RaftConfigPB RaftConsensus::CommittedConfig() const {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    return cmeta_->CommittedConfig();
  }
  
  void RaftConsensus::DumpStatusHtml(std::ostream& out) const {
    RaftPeerPB::Role role;
    {
      ThreadRestrictions::AssertWaitAllowed();
      LockGuard l(lock_);
      if (state_ != kRunning) {
        out << "Tablet " << EscapeForHtmlToString(tablet_id()) << " not running" << std::endl;
        return;
      }
      role = cmeta_->active_role();
    }
  
    out << "<h1>Raft Consensus State</h1>" << std::endl;
  
    out << "<h2>State</h2>" << std::endl;
    out << "<pre>" << EscapeForHtmlToString(ToString()) << "</pre>" << std::endl;
    out << "<h2>Queue</h2>" << std::endl;
    out << "<pre>" << EscapeForHtmlToString(queue_->ToString()) << "</pre>" << std::endl;
  
    // Dump the queues on a leader.
    if (role == RaftPeerPB::LEADER) {
      out << "<h2>Queue overview</h2>" << std::endl;
      out << "<pre>" << EscapeForHtmlToString(queue_->ToString()) << "</pre>" << std::endl;
      out << "<hr/>" << std::endl;
      out << "<h2>Queue details</h2>" << std::endl;
      queue_->DumpToHtml(out);
    }
  }
  
  void RaftConsensus::ElectionCallback(ElectionReason reason, const ElectionResult& result) {
    // We're running on a reactor thread; service the callback on another thread.
    //
    // There's no need to reenable the failure detector; if this fails, it's a
    // sign that RaftConsensus has stopped and we no longer need failure detection.
    auto self = shared_from_this();
    Status s = raft_pool_token_->Submit([=]() { self->DoElectionCallback(reason, result); });
    if (!s.ok()) {
      static const char* msg = "unable to run election callback";
      CHECK(s.IsServiceUnavailable()) << LogPrefixThreadSafe() << msg;
      WARN_NOT_OK(s, LogPrefixThreadSafe() + msg);
    }
  }
  
  void RaftConsensus::DoElectionCallback(ElectionReason reason, const ElectionResult& result) {
    const int64_t election_term = result.vote_request.candidate_term();
    const bool was_pre_election = result.vote_request.is_pre_election();
    const char* election_type = was_pre_election ? "pre-election" : "election";
  
    ThreadRestrictions::AssertWaitAllowed();
    UniqueLock l(lock_);
  
    Status s = CheckRunningUnlocked();
    if (PREDICT_FALSE(!s.ok())) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Received " << election_type << " callback for term "
                                     << election_term << " while not running: "
                                     << s.ToString();
      return;
    }
  
    // If this election was not triggered by the failure detector, the fd may
    // still be enabled and needs to be snoozed, both if we win and lose:
    // - When we win because we're about to disable it and become leader.
    // - When we lose or otherwise we can fall into a cycle, where everyone keeps
    //   triggering elections but no election ever completes because by the time they
    //   finish another one is triggered already.
    //
    // This is a no-op if the failure detector is disabled..
    MonoDelta snooze_delta = LeaderElectionExpBackoffDeltaUnlocked();
    SnoozeFailureDetector(string("election complete"), snooze_delta);
    auto enable_fd = MakeScopedCleanup([&]() {
      // The failure detector was disabled if it triggered this election. Reenable
      // it when exiting this function.
      if (reason == ELECTION_TIMEOUT_EXPIRED) {
        EnableFailureDetector(snooze_delta);
      }
    });
  
    if (result.decision == VOTE_DENIED) {
      failed_elections_since_stable_leader_++;
      num_failed_elections_metric_->set_value(failed_elections_since_stable_leader_);
  
      // If we called an election and one of the voters had a higher term than we did,
      // we should bump our term before we potentially try again. This is particularly
      // important with pre-elections to avoid getting "stuck" in a case like:
      //    Peer A: has ops through 1.10, term = 2, voted in term 2 for peer C
      //    Peer B: has ops through 1.15, term = 1
      // In this case, Peer B will reject peer A's pre-elections for term 3 because
      // the local log is longer. Peer A will reject B's pre-elections for term 2
      // because it already voted in term 2. The check below ensures that peer B
      // will bump to term 2 when it gets the vote rejection, such that its
      // next pre-election (for term 3) would succeed.
      if (result.highest_voter_term > CurrentTermUnlocked()) {
        HandleTermAdvanceUnlocked(result.highest_voter_term);
      }
  
      LOG_WITH_PREFIX_UNLOCKED(INFO)
          << "Leader " << election_type << " lost for term " << election_term
          << ". Reason: "
          << (!result.message.empty() ? result.message : "None given");
      return;
    }
  
    // In a pre-election, we collected votes for the _next_ term.
    // So, we need to adjust our expectations of what the current term should be.
    int64_t election_started_in_term = election_term;
    if (was_pre_election) {
      election_started_in_term--;
    }
  
    if (election_started_in_term != CurrentTermUnlocked()) {
      LOG_WITH_PREFIX_UNLOCKED(INFO)
          << "Leader " << election_type << " decision vote started in "
          << "defunct term " << election_started_in_term << ": won";
      return;
    }
  
    if (!cmeta_->IsVoterInConfig(peer_uuid(), ACTIVE_CONFIG)) {
      LOG_WITH_PREFIX_UNLOCKED(WARNING) << "Leader " << election_type
                                        << " decision while not in active config. "
                                        << "Result: Term " << election_term
                                        << ": won. RaftConfig: "
                                        << SecureShortDebugString(cmeta_->ActiveConfig());
      return;
    }
  
    // At this point we're either already leader, we're going to become leader,
    // or we're going to run a real election.
    //
    // In all cases, do not reenable the failure detector.
    enable_fd.cancel();
  
    if (cmeta_->active_role() == RaftPeerPB::LEADER) {
      // If this was a pre-election, it's possible to see the following interleaving:
      //
      //  1. Term N (follower): send a real election for term N.
      //  2. An externally-triggered election is started.
      //  3. Term N (follower): send a pre-election for term N+1.
      //  4. Election callback for real election from term N completes.
      //     Peer is now leader for term N.
      //  5. Pre-election callback from term N+1 completes, even though
      //     we are currently a leader of term N.
      // In this case, we should just ignore the pre-election, since we're
      // happily the leader of the prior term.
      if (was_pre_election) return;
      LOG_WITH_PREFIX_UNLOCKED(DFATAL)
          << "Leader " << election_type << " callback while already leader! "
          << "Result: Term " << election_term << ": won";
      return;
    }
  
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Leader " << election_type << " won for term " << election_term;
  
    if (was_pre_election) {
      // We just won the pre-election. So, we need to call a real election.
      l.unlock();
      WARN_NOT_OK_EVERY_N_SECS(StartElection(NORMAL_ELECTION, reason),
                               "Couldn't start leader election after successful pre-election", 10);
    } else {
      // We won a real election. Convert role to LEADER.
      SetLeaderUuidUnlocked(peer_uuid());
  
      // TODO(todd): BecomeLeaderUnlocked() can fail due to state checks during shutdown.
      // It races with the above state check.
      // This could be a problem during tablet deletion.
      CHECK_OK(BecomeLeaderUnlocked());
    }
  }
  
  boost::optional<OpId> RaftConsensus::GetLastOpId(OpIdType type) {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    return GetLastOpIdUnlocked(type);
  }
  
  boost::optional<OpId> RaftConsensus::GetLastOpIdUnlocked(OpIdType type) {
    // Return early if this method is called on an instance of RaftConsensus that
    // has not yet been started, failed during Init(), or failed during Start().
    if (!queue_ || !pending_) return boost::none;
  
    switch (type) {
      case RECEIVED_OPID:
        return queue_->GetLastOpIdInLog();
      case COMMITTED_OPID:
        return MakeOpId(pending_->GetTermWithLastCommittedOp(),
                        pending_->GetCommittedIndex());
      default:
        LOG(DFATAL) << LogPrefixUnlocked() << "Invalid OpIdType " << type;
        return boost::none;
    }
  }
  
  log::RetentionIndexes RaftConsensus::GetRetentionIndexes() {
    // Grab the watermarks from the queue. It's OK to fetch these two watermarks
    // separately -- the worst case is we see a relatively "out of date" watermark
    // which just means we'll retain slightly more than necessary in this invocation
    // of log GC.
    return log::RetentionIndexes(queue_->GetCommittedIndex(), // for durability
                                 queue_->GetAllReplicatedIndex()); // for peers
  }
  
  void RaftConsensus::MarkDirty(const string& reason) {
    WARN_NOT_OK(raft_pool_token_->Submit([=]() { this->mark_dirty_clbk_(reason); }),
                LogPrefixThreadSafe() + "Unable to run MarkDirty callback");
  }
  
  void RaftConsensus::MarkDirtyOnSuccess(const string& reason,
                                         const StatusCallback& client_cb,
                                         const Status& status) {
    if (PREDICT_TRUE(status.ok())) {
      MarkDirty(reason);
    }
    client_cb(status);
  }
  
  void RaftConsensus::NonTxRoundReplicationFinished(ConsensusRound* round,
                                                    const StatusCallback& client_cb,
                                                    const Status& status) {
    // NOTE: lock_ is held here because this is triggered by
    // PendingRounds::AbortOpsAfter() and AdvanceCommittedIndex().
    DCHECK(lock_.is_locked());
    OperationType op_type = round->replicate_msg()->op_type();
    const string& op_type_str = OperationType_Name(op_type);
    CHECK(IsConsensusOnlyOperation(op_type)) << "Unexpected op type: " << op_type_str;
  
    if (op_type == CHANGE_CONFIG_OP) {
      CompleteConfigChangeRoundUnlocked(round, status);
      // Fall through to the generic handling.
    }
  
    // TODO(mpercy): May need some refactoring to unlock 'lock_' before invoking
    // the client callback.
  
    if (!status.ok()) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << op_type_str << " replication failed: "
                                     << status.ToString();
      client_cb(status);
      return;
    }
    VLOG_WITH_PREFIX_UNLOCKED(1) << "Committing " << op_type_str << " with op id "
                                 << round->id();
    round_handler_->FinishConsensusOnlyRound(round);
    unique_ptr<CommitMsg> commit_msg(new CommitMsg);
    commit_msg->set_op_type(round->replicate_msg()->op_type());
    *commit_msg->mutable_commited_op_id() = round->id();
  
    CHECK_OK(log_->AsyncAppendCommit(
        std::move(commit_msg), [](const Status& s) {
          CrashIfNotOkStatusCB("Enqueued commit operation failed to write to WAL", s);
        }));
  
    client_cb(status);
  }
  
  void RaftConsensus::CompleteConfigChangeRoundUnlocked(ConsensusRound* round, const Status& status) {
    DCHECK(lock_.is_locked());
    const OpId& op_id = round->replicate_msg()->id();
  
    if (!status.ok()) {
      // If the config change being aborted is the current pending one, abort it.
      if (cmeta_->has_pending_config() &&
          cmeta_->GetConfigOpIdIndex(PENDING_CONFIG) == op_id.index()) {
        LOG_WITH_PREFIX_UNLOCKED(INFO) << "Aborting config change with OpId "
                                       << op_id << ": " << status.ToString();
        cmeta_->clear_pending_config();
  
        // Disable leader failure detection if transitioning from VOTER to
        // NON_VOTER and vice versa.
        UpdateFailureDetectorState();
      } else {
        LOG_WITH_PREFIX_UNLOCKED(INFO)
            << "Skipping abort of non-pending config change with OpId "
            << op_id << ": " << status.ToString();
      }
  
      // It's possible to abort a config change which isn't the pending one in the following
      // sequence:
      // - replicate a config change
      // - it gets committed, so we write the new config to disk as the Committed configuration
      // - we crash before the COMMIT message hits the WAL
      // - we restart the server, and the config change is added as a pending round again,
      //   but isn't set as Pending because it's already committed.
      // - we delete the tablet before committing it
      // See KUDU-1735.
      return;
    }
  
    // Commit the successful config change.
  
    DCHECK(round->replicate_msg()->change_config_record().has_old_config());
    DCHECK(round->replicate_msg()->change_config_record().has_new_config());
    const RaftConfigPB& old_config = round->replicate_msg()->change_config_record().old_config();
    const RaftConfigPB& new_config = round->replicate_msg()->change_config_record().new_config();
    DCHECK(old_config.has_opid_index());
    DCHECK(new_config.has_opid_index());
    // Check if the pending Raft config has an OpId less than the committed
    // config. If so, this is a replay at startup in which the COMMIT
    // messages were delayed.
    int64_t committed_config_opid_index = cmeta_->GetConfigOpIdIndex(COMMITTED_CONFIG);
    if (new_config.opid_index() > committed_config_opid_index) {
      LOG_WITH_PREFIX_UNLOCKED(INFO)
          << "Committing config change with OpId "
          << op_id << ": "
          << DiffRaftConfigs(old_config, new_config)
          << ". New config: { " << SecureShortDebugString(new_config) << " }";
      CHECK_OK(SetCommittedConfigUnlocked(new_config));
    } else {
      LOG_WITH_PREFIX_UNLOCKED(INFO)
          << "Ignoring commit of config change with OpId "
          << op_id << " because the committed config has OpId index "
          << committed_config_opid_index << ". The config change we are ignoring is: "
          << "Old config: { " << SecureShortDebugString(old_config) << " }. "
          << "New config: { " << SecureShortDebugString(new_config) << " }";
    }
  }
  
  void RaftConsensus::EnableFailureDetector(boost::optional<MonoDelta> delta) {
    if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
      failure_detector_->Start(std::move(delta));
    }
  }
  
  void RaftConsensus::DisableFailureDetector() {
    if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
      failure_detector_->Stop();
    }
  }
  
  void RaftConsensus::UpdateFailureDetectorState(boost::optional<MonoDelta> delta) {
    DCHECK(lock_.is_locked());
    const auto& uuid = peer_uuid();
    if (uuid != cmeta_->leader_uuid() &&
        cmeta_->IsVoterInConfig(uuid, ACTIVE_CONFIG)) {
      // A voter that is not the leader should run the failure detector.
      EnableFailureDetector(std::move(delta));
    } else {
      // Otherwise, the local peer should not start leader elections
      // (e.g. if it is the leader, a non-voter, a non-participant, etc).
      DisableFailureDetector();
    }
  }
  
  void RaftConsensus::SnoozeFailureDetector(boost::optional<string> reason_for_log,
                                            boost::optional<MonoDelta> delta) {
    if (PREDICT_TRUE(failure_detector_ && FLAGS_enable_leader_failure_detection)) {
      if (reason_for_log) {
        VLOG(1) << LogPrefixThreadSafe()
                << Substitute("Snoozing failure detection for $0 ($1)",
                              delta ? delta->ToString() : "election timeout",
                              *reason_for_log);
      }
  
      if (!delta) {
        delta = MinimumElectionTimeout();
      }
      failure_detector_->Snooze(std::move(delta));
    }
  }
  
  void RaftConsensus::WithholdVotes() {
    MonoTime prev = withhold_votes_until_;
    MonoTime next = MonoTime::Now() + MinimumElectionTimeout();
    do {
      if (prev == MonoTime::Max()) {
        // Maximum withholding time already. It might be the case if replica
        // has become a leader already.
        break;
      }
      next = MonoTime::Now() + MinimumElectionTimeout();
    } while (!withhold_votes_until_.compare_exchange_weak(prev, next));
  }
  
  MonoDelta RaftConsensus::LeaderElectionExpBackoffDeltaUnlocked() {
    DCHECK(lock_.is_locked());
    // Compute a backoff factor based on how many leader elections have
    // failed since a stable leader was last seen.
    double backoff_factor = pow(1.5, failed_elections_since_stable_leader_ + 1);
    double min_timeout = MinimumElectionTimeout().ToMilliseconds();
    double max_timeout = std::min<double>(
        min_timeout * backoff_factor,
        FLAGS_leader_failure_exp_backoff_max_delta_ms);
  
    // Randomize the timeout between the minimum and the calculated value.
    // We do this after the above capping to the max. Otherwise, after a
    // churny period, we'd end up highly likely to backoff exactly the max
    // amount.
    double timeout = min_timeout + (max_timeout - min_timeout) * rng_.NextDoubleFraction();
    DCHECK_GE(timeout, min_timeout);
  
    return MonoDelta::FromMilliseconds(timeout);
  }
  
  Status RaftConsensus::HandleTermAdvanceUnlocked(ConsensusTerm new_term,
                                                  FlushToDisk flush) {
    DCHECK(lock_.is_locked());
    if (new_term <= CurrentTermUnlocked()) {
      return Status::IllegalState(Substitute("Can't advance term to: $0 current term: $1 is higher.",
                                             new_term, CurrentTermUnlocked()));
    }
    if (cmeta_->active_role() == RaftPeerPB::LEADER) {
      LOG_WITH_PREFIX_UNLOCKED(INFO) << "Stepping down as leader of term "
                                     << CurrentTermUnlocked();
      RETURN_NOT_OK(BecomeReplicaUnlocked());
    }
  
    LOG_WITH_PREFIX_UNLOCKED(INFO) << "Advancing to term " << new_term;
    RETURN_NOT_OK(SetCurrentTermUnlocked(new_term, flush));
    if (term_metric_) term_metric_->set_value(new_term);
    last_received_cur_leader_ = MinimumOpId();
    return Status::OK();
  }
  
  Status RaftConsensus::CheckSafeToReplicateUnlocked(const ReplicateMsg& msg) const {
    DCHECK(lock_.is_locked());
    DCHECK(!msg.has_id()) << "Should not have an ID yet: " << SecureShortDebugString(msg);
    RETURN_NOT_OK(CheckRunningUnlocked());
    return CheckActiveLeaderUnlocked();
  }
  
  Status RaftConsensus::CheckRunningUnlocked() const {
    DCHECK(lock_.is_locked());
    if (PREDICT_FALSE(state_ != kRunning)) {
      return Status::IllegalState("RaftConsensus is not running",
                                  Substitute("State = $0", State_Name(state_)));
    }
    return Status::OK();
  }
  
  Status RaftConsensus::CheckActiveLeaderUnlocked() const {
    DCHECK(lock_.is_locked());
    RaftPeerPB::Role role = cmeta_->active_role();
    switch (role) {
      case RaftPeerPB::LEADER:
        // Check for the consistency of the information in the consensus metadata
        // and the state of the consensus queue.
        DCHECK(queue_->IsInLeaderMode());
        if (leader_transfer_in_progress_.Load()) {
          return Status::ServiceUnavailable("leader transfer in progress");
        }
        return Status::OK();
  
      default:
        // Check for the consistency of the information in the consensus metadata
        // and the state of the consensus queue.
        DCHECK(!queue_->IsInLeaderMode());
        return Status::IllegalState(Substitute("Replica $0 is not leader of this config. Role: $1. "
                                               "Consensus state: $2",
                                               peer_uuid(),
                                               RaftPeerPB::Role_Name(role),
                                               SecureShortDebugString(cmeta_->ToConsensusStatePB())));
    }
  }
  
  Status RaftConsensus::CheckNoConfigChangePendingUnlocked() const {
    DCHECK(lock_.is_locked());
    if (cmeta_->has_pending_config()) {
      return Status::IllegalState(
          Substitute("RaftConfig change currently pending. Only one is allowed at a time.\n"
                     "  Committed config: $0.\n  Pending config: $1",
                     SecureShortDebugString(cmeta_->CommittedConfig()),
                     SecureShortDebugString(cmeta_->PendingConfig())));
    }
    return Status::OK();
  }
  
  Status RaftConsensus::SetPendingConfigUnlocked(const RaftConfigPB& new_config) {
    DCHECK(lock_.is_locked());
    RETURN_NOT_OK_PREPEND(VerifyRaftConfig(new_config),
                          "Invalid config to set as pending");
    if (!new_config.unsafe_config_change()) {
      CHECK(!cmeta_->has_pending_config())
          << "Attempt to set pending config while another is already pending! "
          << "Existing pending config: " << SecureShortDebugString(cmeta_->PendingConfig()) << "; "
          << "Attempted new pending config: " << SecureShortDebugString(new_config);
    } else if (cmeta_->has_pending_config()) {
      LOG_WITH_PREFIX_UNLOCKED(INFO)
          << "Allowing unsafe config change even though there is a pending config! "
          << "Existing pending config: " << SecureShortDebugString(cmeta_->PendingConfig()) << "; "
          << "New pending config: " << SecureShortDebugString(new_config);
    }
    cmeta_->set_pending_config(new_config);
  
    UpdateFailureDetectorState();
  
    return Status::OK();
  }
  
  Status RaftConsensus::SetCommittedConfigUnlocked(const RaftConfigPB& config_to_commit) {
    TRACE_EVENT0("consensus", "RaftConsensus::SetCommittedConfigUnlocked");
    DCHECK(lock_.is_locked());
    DCHECK(config_to_commit.IsInitialized());
    RETURN_NOT_OK_PREPEND(VerifyRaftConfig(config_to_commit),
                          "Invalid config to set as committed");
  
    // Compare committed with pending configuration, ensure that they are the same.
    // In the event of an unsafe config change triggered by an administrator,
    // it is possible that the config being committed may not match the pending config
    // because unsafe config change allows multiple pending configs to exist.
    // Therefore we only need to validate that 'config_to_commit' matches the pending config
    // if the pending config does not have its 'unsafe_config_change' flag set.
    if (cmeta_->has_pending_config()) {
      RaftConfigPB pending_config = cmeta_->PendingConfig();
      if (!pending_config.unsafe_config_change()) {
        // Quorums must be exactly equal, even w.r.t. peer ordering.
        CHECK_EQ(pending_config.SerializeAsString(),
                 config_to_commit.SerializeAsString())
            << Substitute("New committed config must equal pending config, but does not. "
                          "Pending config: $0, committed config: $1",
                          SecureShortDebugString(pending_config),
                          SecureShortDebugString(config_to_commit));
      }
    }
    cmeta_->set_committed_config(config_to_commit);
    cmeta_->clear_pending_config();
    CHECK_OK(cmeta_->Flush());
    return Status::OK();
  }
  
  Status RaftConsensus::SetCurrentTermUnlocked(int64_t new_term,
                                              FlushToDisk flush) {
    TRACE_EVENT1("consensus", "RaftConsensus::SetCurrentTermUnlocked",
                 "term", new_term);
    DCHECK(lock_.is_locked());
    if (PREDICT_FALSE(new_term <= CurrentTermUnlocked())) {
      return Status::IllegalState(
          Substitute("Cannot change term to a term that is lower than or equal to the current one. "
                     "Current: $0, Proposed: $1", CurrentTermUnlocked(), new_term));
    }
    cmeta_->set_current_term(new_term);
    cmeta_->clear_voted_for();
    if (flush == FLUSH_TO_DISK) {
      CHECK_OK(cmeta_->Flush());
    }
    ClearLeaderUnlocked();
    return Status::OK();
  }
  
  int64_t RaftConsensus::CurrentTermUnlocked() const {
    DCHECK(lock_.is_locked());
    return cmeta_->current_term();
  }
  
  string RaftConsensus::GetLeaderUuidUnlocked() const {
    DCHECK(lock_.is_locked());
    return cmeta_->leader_uuid();
  }
  
  bool RaftConsensus::HasLeaderUnlocked() const {
    DCHECK(lock_.is_locked());
    return !GetLeaderUuidUnlocked().empty();
  }
  
  void RaftConsensus::ClearLeaderUnlocked() {
    DCHECK(lock_.is_locked());
    cmeta_->set_leader_uuid("");
  }
  
  bool RaftConsensus::HasVotedCurrentTermUnlocked() const {
    DCHECK(lock_.is_locked());
    return cmeta_->has_voted_for();
  }
  
  Status RaftConsensus::SetVotedForCurrentTermUnlocked(const string& uuid) {
    TRACE_EVENT1("consensus", "RaftConsensus::SetVotedForCurrentTermUnlocked",
                 "uuid", uuid);
    DCHECK(lock_.is_locked());
    cmeta_->set_voted_for(uuid);
    CHECK_OK(cmeta_->Flush());
    return Status::OK();
  }
  
  const string& RaftConsensus::GetVotedForCurrentTermUnlocked() const {
    DCHECK(lock_.is_locked());
    DCHECK(cmeta_->has_voted_for());
    return cmeta_->voted_for();
  }
  
  const ConsensusOptions& RaftConsensus::GetOptions() const {
    return options_;
  }
  
  string RaftConsensus::LogPrefix() const {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    return LogPrefixUnlocked();
  }
  
  string RaftConsensus::LogPrefixUnlocked() const {
    DCHECK(lock_.is_locked());
    // 'cmeta_' may not be set if initialization failed.
    string cmeta_info;
    if (cmeta_) {
      cmeta_info = Substitute(" [term $0 $1]",
                              cmeta_->current_term(),
                              RaftPeerPB::Role_Name(cmeta_->active_role()));
    }
    return Substitute("T $0 P $1$2: ", options_.tablet_id, peer_uuid(), cmeta_info);
  }
  
  string RaftConsensus::LogPrefixThreadSafe() const {
    return Substitute("T $0 P $1: ",
                      options_.tablet_id,
                      peer_uuid());
  }
  
  string RaftConsensus::ToString() const {
    ThreadRestrictions::AssertWaitAllowed();
    LockGuard l(lock_);
    return ToStringUnlocked();
  }
  
  string RaftConsensus::ToStringUnlocked() const {
    DCHECK(lock_.is_locked());
    return Substitute("Replica: $0, State: $1, Role: $2",
                      peer_uuid(), State_Name(state_), RaftPeerPB::Role_Name(cmeta_->active_role()));
  }
  
  int64_t RaftConsensus::MetadataOnDiskSize() const {
    return cmeta_->on_disk_size();
  }
  
  ConsensusMetadata* RaftConsensus::consensus_metadata_for_tests() const {
    return cmeta_.get();
  }
  
  int64_t RaftConsensus::GetMillisSinceLastLeaderHeartbeat() const {
      return last_leader_communication_time_micros_ == 0 ?
          0 : (GetMonoTimeMicros() - last_leader_communication_time_micros_) / 1000;
  }
  
  ////////////////////////////////////////////////////////////////////////
  // ConsensusBootstrapInfo
  ////////////////////////////////////////////////////////////////////////
  
  ConsensusBootstrapInfo::ConsensusBootstrapInfo()
    : last_id(MinimumOpId()),
      last_committed_id(MinimumOpId()) {
  }
  
  ConsensusBootstrapInfo::~ConsensusBootstrapInfo() {
    STLDeleteElements(&orphaned_replicates);
  }
  
  ////////////////////////////////////////////////////////////////////////
  // ConsensusRound
  ////////////////////////////////////////////////////////////////////////
  
  ConsensusRound::ConsensusRound(RaftConsensus* consensus,
                                 unique_ptr<ReplicateMsg> replicate_msg,
                                 ConsensusReplicatedCallback replicated_cb)
      : consensus_(consensus),
        replicate_msg_(new RefCountedReplicate(replicate_msg.release())),
        replicated_cb_(std::move(replicated_cb)),
        bound_term_(-1) {}
  
  ConsensusRound::ConsensusRound(RaftConsensus* consensus,
                                 ReplicateRefPtr replicate_msg)
      : consensus_(consensus),
        replicate_msg_(std::move(replicate_msg)),
        bound_term_(-1) {
    DCHECK(replicate_msg_);
  }
  
  void ConsensusRound::NotifyReplicationFinished(const Status& status) {
    if (PREDICT_FALSE(!replicated_cb_)) return;
    replicated_cb_(status);
  }
  
  Status ConsensusRound::CheckBoundTerm(int64_t current_term) const {
    if (PREDICT_FALSE(bound_term_ != -1 &&
                      bound_term_ != current_term)) {
      return Status::Aborted(
        strings::Substitute(
          "Op submitted in term $0 cannot be replicated in term $1",
          bound_term_, current_term));
    }
    return Status::OK();
  }
  
  }  // namespace consensus
  }  // namespace kudu
  