// 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 <cstdint>
#include <functional>
#include <initializer_list>
#include <memory>
#include <optional>
#include <ostream>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include <gflags/gflags_declare.h>
#include <glog/logging.h>
#include <gtest/gtest.h>

#include "kudu/common/common.pb.h"
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/consensus.pb.h"
#include "kudu/consensus/metadata.pb.h"
#include "kudu/consensus/quorum_util.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/integration-tests/cluster_itest_util.h"
#include "kudu/integration-tests/external_mini_cluster-itest-base.h"
#include "kudu/integration-tests/mini_cluster_fs_inspector.h"
#include "kudu/integration-tests/raft_consensus-itest-base.h"
#include "kudu/integration-tests/test_workload.h"
#include "kudu/master/master.pb.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/tserver/tablet_server-test-base.h"
#include "kudu/util/monotime.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"

using kudu::consensus::ADD_PEER;
using kudu::consensus::COMMITTED_OPID;
using kudu::consensus::ConsensusStatePB;
using kudu::consensus::GetRaftConfigMember;
using kudu::consensus::MODIFY_PEER;
using kudu::consensus::RaftPeerAttrsPB;
using kudu::consensus::RaftPeerPB;
using kudu::consensus::REMOVE_PEER;
using kudu::consensus::EXCLUDE_HEALTH_REPORT;
using kudu::itest::BulkChangeConfig;
using kudu::itest::GetTableLocations;
using kudu::itest::TServerDetails;
using kudu::itest::GetConsensusState;
using kudu::tserver::RaftConsensusITestBase;
using kudu::master::VOTER_REPLICA;
using kudu::pb_util::SecureShortDebugString;
using std::nullopt;
using std::optional;
using std::string;
using std::unordered_set;
using std::vector;
using strings::Substitute;

DECLARE_int32(num_replicas);
DECLARE_int32(num_tablet_servers);

namespace kudu {

class RaftConfigChangeITest : public ExternalMiniClusterITestBase {
};

// Regression test for KUDU-2147. In this test we cause an initial leader to
// heartbeat to a master with a new configuration change immediately after it
// has lost its leadership, reporting a new configuration with no leader. The
// master should update its record of which replica is the leader after a new
// leader is elected in that term.
//
// Steps followed in this test:
//
// 1. Inject latency into TS heartbeats to the master and reduce Raft leader
//    heartbeat intervals to followers to be able to carefully control the
//    sequence of events in this test.
// 2. Evict a follower from the config.
// 3. Immediately start an election on the remaining follower in the config.
//    We know that this follower should win the election because in order to
//    commit a removal from 3 voters, the removal op must be replicated to both
//    of the two remaining voters in the config.
// 4. The master will get the config change heartbeat from the initial leader,
//    which will indicate a new term and no current leader.
// 5. The master will then get a heartbeat from the new leader to report that
//    it's now the leader of the config.
// 6. Once the master knows who the new leader is, the master will instruct the
//    new leader to add a new replica to its config to bring it back up to 3
//    voters. That new replica will be the follower that we evicted in step 2.
// 7. If that succeeds then the leader will replicate the eviction, the
//    leader's own no-op, and the adding-back-in of the evicted replica to that
//    evicted replica's log.
// 8. Once that process completes, all 3 replicas will have identical logs,
//    which is what we wait for at the end of the test.
TEST_F(RaftConfigChangeITest, TestKudu2147) {
  SKIP_IF_SLOW_NOT_ALLOWED();
  const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
  // Slow down leader heartbeats so that in the explicit election below, the
  // second leader does not immediately heartbeat to the initial leader. If
  // that occurs, the initial leader will not report to the master that there
  // is currently no leader.
  NO_FATALS(StartCluster({"--raft_heartbeat_interval_ms=3000"}));

  // Create a new table.
  TestWorkload workload(cluster_.get());
  workload.Setup();
  workload.Start();
  ASSERT_EVENTUALLY([&] {
    ASSERT_GE(workload.batches_completed(), 10);
  });
  workload.StopAndJoin();

  // The table should have replicas on three tservers.
  ASSERT_OK(inspect_->WaitForReplicaCount(3));
  master::GetTableLocationsResponsePB table_locations;
  ASSERT_OK(GetTableLocations(cluster_->master_proxy(), TestWorkload::kDefaultTableName,
                              kTimeout, VOTER_REPLICA, /*table_id=*/nullopt, &table_locations));
  ASSERT_EQ(1, table_locations.tablet_locations_size()); // Only 1 tablet.
  ASSERT_EQ(3, table_locations.tablet_locations(0).interned_replicas_size()); // 3 replicas.
  string tablet_id = table_locations.tablet_locations(0).tablet_id();

  // Wait for all 3 replicas to converge before we start the test.
  ASSERT_OK(WaitForServersToAgree(kTimeout, ts_map_, tablet_id, 1));

  // Retry as needed to counter normal leader election activity.
  ASSERT_EVENTUALLY([&] {
    // Find initial leader.
    TServerDetails* leader;
    ASSERT_OK(FindTabletLeader(ts_map_, tablet_id, kTimeout, &leader));
    ASSERT_OK(WaitForOpFromCurrentTerm(leader, tablet_id, COMMITTED_OPID, kTimeout));

    vector<TServerDetails*> followers;
    for (int i = 0; i < cluster_->num_tablet_servers(); i++) {
      // Dynamically set the latency injection flag to induce TS -> master
      // heartbeat delays. The leader delays its master heartbeats by 2 sec
      // each time; followers delay their master heartbeats by even longer.
      int ms_delay = cluster_->tablet_server(i)->uuid() == leader->uuid() ? 2000 : 5000;
      ASSERT_OK(cluster_->SetFlag(cluster_->tablet_server(i),
                "heartbeat_inject_latency_before_heartbeat_ms",
                Substitute("$0", ms_delay)));
      // Keep track of the followers.
      if (cluster_->tablet_server(i)->uuid() != leader->uuid()) {
        followers.push_back(ts_map_[cluster_->tablet_server(i)->uuid()]);
      }
    }
    ASSERT_EQ(2, followers.size());

    // Now that heartbeat injection is enabled, evict one follower and trigger
    // an election on another follower immediately thereafter.
    ASSERT_OK(itest::RemoveServer(leader, tablet_id, followers[0], kTimeout));

    // Immediately start an election on the remaining follower. This will cause
    // the initial leader's term to rev and it will have to step down. When it
    // sends a tablet report to the master with the new configuration excluding
    // the removed tablet it will report an unknown leader in the new term.
    ASSERT_OK(itest::StartElection(followers[1], tablet_id, kTimeout));
  });

  // Wait until the master re-adds the evicted replica and it is back up and
  // running. If the master hit KUDU-2147, this would fail because the master
  // would be unable to add the removed server back, and that replica would be
  // missing the config change op that removed it from the config.
  ASSERT_OK(WaitForServersToAgree(kTimeout, ts_map_, tablet_id, 1));
}

// Test automatic promotion of a non-voter replica in a 3-4-3 re-replication
// (KUDU-1097) paradigm.
TEST_F(RaftConfigChangeITest, TestNonVoterPromotion) {
  const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
  // Enable 3-4-3 re-replication.
  NO_FATALS(StartCluster({"--raft_prepare_replacement_before_eviction=true"},
                         {"--raft_prepare_replacement_before_eviction=true",
                          "--catalog_manager_evict_excess_replicas=false"},
                         /*num_tablet_servers=*/ 4));
  TestWorkload workload(cluster_.get());
  workload.Setup();

  // The table should initially have replicas on three tservers.
  ASSERT_OK(inspect_->WaitForReplicaCount(3));
  master::GetTableLocationsResponsePB table_locations;
  ASSERT_OK(GetTableLocations(cluster_->master_proxy(), TestWorkload::kDefaultTableName,
                              kTimeout, VOTER_REPLICA, /*table_id=*/nullopt, &table_locations));
  ASSERT_EQ(1, table_locations.tablet_locations_size()); // Only 1 tablet.
  ASSERT_EQ(3, table_locations.tablet_locations(0).interned_replicas_size()); // 3 replicas.
  string tablet_id = table_locations.tablet_locations(0).tablet_id();

  // Find the TS that does not have a replica.
  unordered_set<string> initial_replicas;
  for (const auto& replica : table_locations.tablet_locations(0).interned_replicas()) {
    const auto& uuid = table_locations.ts_infos(replica.ts_info_idx()).permanent_uuid();
    initial_replicas.insert(uuid);
  }
  TServerDetails* new_replica = nullptr;
  for (const auto& entry : ts_map_) {
    if (!ContainsKey(initial_replicas, entry.first)) {
      new_replica = entry.second;
      break;
    }
  }
  ASSERT_NE(nullptr, new_replica);

  TServerDetails* leader_replica = nullptr;
  ASSERT_EVENTUALLY([&] {
    ASSERT_OK(FindTabletLeader(ts_map_, tablet_id, kTimeout, &leader_replica));
    ASSERT_NE(new_replica, leader_replica);

    // Add the 4th replica as a NON_VOTER with promote=true.
    RaftPeerAttrsPB attrs;
    attrs.set_promote(true);
    ASSERT_OK(AddServer(leader_replica, tablet_id, new_replica,
                        RaftPeerPB::NON_VOTER, kTimeout, attrs));
  });

  // Wait for there to be 4 voters in the config.
  ASSERT_OK(WaitUntilCommittedConfigNumVotersIs(/*num_voters=*/ 4,
                                                leader_replica,
                                                tablet_id,
                                                kTimeout));

  NO_FATALS(cluster_->AssertNoCrashes());
}

// Functional test for the BulkChangeConfig RPC API.
TEST_F(RaftConfigChangeITest, TestBulkChangeConfig) {
  const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
  const int kNumTabletServers = 4;
  const int kNumInitialReplicas = 3;
  NO_FATALS(StartCluster({"--enable_leader_failure_detection=false",
                          "--raft_prepare_replacement_before_eviction=true"},
                         {"--catalog_manager_wait_for_new_tablets_to_elect_leader=false",
                          "--raft_prepare_replacement_before_eviction=true"},
                         kNumTabletServers));

  // Create a table.
  TestWorkload workload(cluster_.get());
  workload.Setup();

  ASSERT_OK(inspect_->WaitForReplicaCount(kNumInitialReplicas));
  master::GetTableLocationsResponsePB table_locations;
  ASSERT_OK(GetTableLocations(cluster_->master_proxy(), TestWorkload::kDefaultTableName,
                              kTimeout, VOTER_REPLICA, /*table_id=*/nullopt, &table_locations));
  ASSERT_EQ(1, table_locations.tablet_locations_size()); // Only 1 tablet.
  ASSERT_EQ(kNumInitialReplicas, table_locations.tablet_locations(0).interned_replicas_size());
  string tablet_id = table_locations.tablet_locations(0).tablet_id();
  unordered_set<int> replica_indexes;
  for (const auto& replica : table_locations.tablet_locations(0).interned_replicas()) {
    const auto& uuid = table_locations.ts_infos(replica.ts_info_idx()).permanent_uuid();
    int idx = cluster_->tablet_server_index_by_uuid(uuid);
    replica_indexes.emplace(idx);
    ASSERT_OK(itest::WaitUntilTabletRunning(ts_map_[cluster_->tablet_server(idx)->uuid()],
                                            tablet_id, kTimeout));
  }
  ASSERT_EQ(kNumInitialReplicas, replica_indexes.size());
  const int kLeaderIndex = *replica_indexes.begin();
  int new_replica_index = -1;
  for (int i = 0; i < kNumTabletServers; i++) {
    if (!ContainsKey(replica_indexes, i)) {
      new_replica_index = i;
    }
  }
  ASSERT_NE(-1, new_replica_index);

  string leader_uuid = cluster_->tablet_server(kLeaderIndex)->uuid();
  auto* leader_replica = ts_map_[leader_uuid];
  ASSERT_OK(itest::StartElection(leader_replica, tablet_id, kTimeout));
  workload.Start();
  while (workload.rows_inserted() < 100) {
    SleepFor(MonoDelta::FromMilliseconds(10));
  }
  workload.StopAndJoin();

  // We don't want the master interfering with the rest of the test.
  cluster_->master()->Shutdown();

  struct BulkSpec {
    consensus::ChangeConfigType change_type;
    int tserver_index;
    RaftPeerPB::MemberType member_type;
    bool replace;
    bool promote;
  };

  // Now comes the actual config change testing.
  auto bulk_change = [&](const vector<BulkSpec>& changes,
                         optional<int64_t> cas_config_index = nullopt) {
    vector<consensus::BulkChangeConfigRequestPB::ConfigChangeItemPB> changes_pb;
    for (const auto& chg : changes) {
      const auto& ts_uuid = cluster_->tablet_server(chg.tserver_index)->uuid();
      auto* replica = ts_map_[ts_uuid];

      consensus::BulkChangeConfigRequestPB::ConfigChangeItemPB change_pb;
      change_pb.set_type(chg.change_type);

      RaftPeerPB* peer = change_pb.mutable_peer();
      peer->set_permanent_uuid(ts_uuid);
      peer->set_member_type(chg.member_type);
      peer->mutable_attrs()->set_replace(chg.replace);
      peer->mutable_attrs()->set_promote(chg.promote);
      *peer->mutable_last_known_addr() = replica->registration.rpc_addresses(0);
      changes_pb.emplace_back(std::move(change_pb));
    }

    LOG(INFO) << "submitting config change with changes:";
    for (const auto& change_pb : changes_pb) {
      LOG(INFO) << SecureShortDebugString(change_pb);
    }
    return BulkChangeConfig(leader_replica, tablet_id, changes_pb,
                            kTimeout, cas_config_index);
  };

  // 1) Add a voter. Change config to: V, V, V, V.
  ASSERT_OK(bulk_change({ { ADD_PEER, new_replica_index, RaftPeerPB::VOTER,
                            /*replace=*/ false, /*promote=*/ false } }));
  ConsensusStatePB cstate;
  ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
  ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size());
  ASSERT_EQ(kNumTabletServers, CountVoters(cstate.committed_config()));

  // 2) Simultaneous voter modification and attribute modification.
  //    Change config to: V, V, N, V+p.
  //    Note: setting a VOTER's attribute promote=true is meaningless.
  int replica1_idx = (kLeaderIndex + 1) % kNumTabletServers;
  int replica2_idx = (kLeaderIndex + 2) % kNumTabletServers;
  ASSERT_OK(bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::NON_VOTER,
                            /*replace=*/ false, /*promote=*/ false },
                          { MODIFY_PEER, replica2_idx, RaftPeerPB::VOTER,
                            /*replace=*/ false,  /*promote=*/ true } }));

  ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
  ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size());
  ASSERT_EQ(kNumInitialReplicas, CountVoters(cstate.committed_config()));

  RaftPeerPB* peer;
  ASSERT_OK(GetRaftConfigMember(cstate.mutable_committed_config(),
                                cluster_->tablet_server(replica2_idx)->uuid(), &peer));
  ASSERT_EQ(RaftPeerPB::VOTER, peer->member_type());
  ASSERT_TRUE(peer->attrs().promote()) << SecureShortDebugString(*peer);

  // 3) Single-attribute modification. Change config to: V, V, N+r, V+p.
  //    Note: at the time of writing, if the master is disabled this
  //    configuration will not trigger any actions such as promotion or
  //    eviction.
  ASSERT_OK(bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::NON_VOTER,
                            /*replace=*/ true, /*promote=*/ false } }));

  ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
  ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size())
      << SecureShortDebugString(cstate);
  ASSERT_EQ(kNumInitialReplicas, CountVoters(cstate.committed_config()))
      << SecureShortDebugString(cstate);

  ASSERT_OK(GetRaftConfigMember(cstate.mutable_committed_config(),
                                cluster_->tablet_server(replica1_idx)->uuid(), &peer));
  ASSERT_EQ(RaftPeerPB::NON_VOTER, peer->member_type());
  ASSERT_TRUE(peer->attrs().replace()) << SecureShortDebugString(*peer);

  // 4) Deny changing config (illegally) from: { V, V, N, V } to: { V, V, V, N }
  //    because that would be both a promotion and a demotion in one step.
  Status s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
                             /*replace=*/ false, /*promote=*/ false },
                           { MODIFY_PEER, replica2_idx, RaftPeerPB::NON_VOTER,
                             /*replace=*/ false, /*promote=*/ false } });
  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
  ASSERT_STR_CONTAINS(s.ToString(), "not safe to modify the VOTER status of "
                                    "more than one peer at a time");

  // 5) The caller must not be allowed to make the leader a NON_VOTER.
  s = bulk_change({ { MODIFY_PEER, kLeaderIndex, RaftPeerPB::NON_VOTER,
                      /*replace=*/ false, /*promote=*/ false } });
  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
  ASSERT_STR_MATCHES(s.ToString(), "Cannot modify member type of peer .* because it is the leader");

  // 6) The 'cas_config_index' flag must be respected, if set.
  int64_t committed_config_opid_index = cstate.committed_config().opid_index();
  s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::NON_VOTER,
                      /*replace=*/ false, /*promote=*/ true } }, committed_config_opid_index + 1);
  ASSERT_TRUE(s.IsIllegalState()) << s.ToString();
  ASSERT_STR_MATCHES(s.ToString(), "specified cas_config_opid_index of .* but "
                                   "the committed config has opid_index of .*");

  // 7) Evict down to 2 voters. We will evict a voter and a non-voter at once.
  ASSERT_OK(bulk_change({ { REMOVE_PEER, replica1_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
                            /*replace=*/ false, /*promote=*/ false },
                          { REMOVE_PEER, replica2_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
                            /*replace=*/ false, /*promote=*/ false } }));
  ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
  ASSERT_EQ(2, cstate.committed_config().peers_size());
  ASSERT_EQ(2, CountVoters(cstate.committed_config()));

  // 8) We should reject adding multiple voters at once.
  s = bulk_change({ { ADD_PEER, replica1_idx, RaftPeerPB::VOTER,
                      /*replace=*/ false, /*promote=*/ false },
                    { ADD_PEER, replica2_idx, RaftPeerPB::VOTER,
                      /*replace=*/ false, /*promote=*/ false } });
  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
  ASSERT_STR_MATCHES(s.ToString(), "not safe to modify the VOTER status of "
                                   "more than one peer at a time");

  // 9) Add them back one at a time so we get to full strength (4 voters) again.
  auto to_restore = { replica1_idx, replica2_idx };
  for (auto r : to_restore) {
    ASSERT_OK(bulk_change({ { ADD_PEER, r, RaftPeerPB::VOTER,
                              /*replace=*/ false, /*promote=*/ false } }));
    ASSERT_OK(WaitUntilNoPendingConfig(leader_replica, tablet_id, kTimeout, &cstate));
  }
  ASSERT_EQ(kNumTabletServers, cstate.committed_config().peers_size());
  ASSERT_EQ(kNumTabletServers, CountVoters(cstate.committed_config()));

  // 10) We should reject removing multiple voters at once.
  s = bulk_change({ { REMOVE_PEER, replica1_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
                      /*replace=*/ false, /*promote=*/ false },
                    { REMOVE_PEER, replica2_idx, RaftPeerPB::UNKNOWN_MEMBER_TYPE,
                      /*replace=*/ false, /*promote=*/ false } });
  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
  ASSERT_STR_MATCHES(s.ToString(), "not safe to modify the VOTER status of "
                                   "more than one peer at a time");

  // 11) Reject no-ops.
  s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
                      /*replace=*/ false, /*promote=*/ false } });
  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
  ASSERT_STR_MATCHES(s.ToString(), "must modify a field when calling MODIFY_PEER");

  // 12) Reject empty bulk change config operations.
  s = bulk_change({ });
  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
  ASSERT_STR_MATCHES(s.ToString(), "requested configuration change does not "
                                   "actually modify the config");

  // 13) Reject multiple changes to the same peer in a single request.
  s = bulk_change({ { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
                      /*replace=*/ true, /*promote=*/ false },
                    { MODIFY_PEER, replica1_idx, RaftPeerPB::VOTER,
                      /*replace=*/ false, /*promote=*/ true } });
  ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
  ASSERT_STR_MATCHES(s.ToString(), "only one change allowed per peer");
}

// KUDU-2800
// Check re-replication during slow tablet replica bootstrap.
class SlowTabletBootstrapTest : public RaftConsensusITestBase {
 protected:
  const MonoDelta kTimeout = MonoDelta::FromSeconds(30);
  // Check that 'added_replica_uuid' was added to the consensus
  // and 'removed_replica_uuid' was removed from consensus.
  void ValidateConsensusStateChanged(const string& added_replica_uuid,
                                     const string& removed_replica_uuid,
                                     const MonoDelta& timeout) {
    TServerDetails* leader = nullptr;
    ASSERT_OK(GetLeaderReplicaWithRetries(tablet_id_, &leader));
    consensus::ConsensusStatePB cstate;
    ASSERT_OK(GetConsensusState(leader, tablet_id_, timeout,
                                EXCLUDE_HEALTH_REPORT, &cstate));
    ASSERT_TRUE(IsRaftConfigMember(added_replica_uuid, cstate.committed_config()));
    ASSERT_FALSE(IsRaftConfigMember(removed_replica_uuid, cstate.committed_config()));
  }
  // Create and start cluster.
  // Returns 'any_replica_sever' - UUID of first server with tablet replica,
  // and 'no_replica_server' - UUID of server without tablet replica.
  void SetUpCluster(string* any_replica_server,
                    string* no_replica_server) {
    vector<string> ts_flags {
      // The default value 5 minutes is very long.
      // So we set timeout 3 seconds in order to quickly
      // remove non-responding replica from consensus
      "--follower_unavailable_considered_failed_sec=3"
    };

    AddFlagsForLogRolls(&ts_flags);
    FLAGS_num_tablet_servers = 4;
    FLAGS_num_replicas = 3;
    NO_FATALS(BuildAndStart(ts_flags, {}));

    ASSERT_EQ(4, tablet_servers_.size());

    // Extra sanity checks.
    vector<string> replica_servers = GetServersWithReplica(tablet_id_);
    ASSERT_EQ(3, replica_servers.size());
    vector<string> no_replica_servers = GetServersWithoutReplica(tablet_id_);
    ASSERT_EQ(1, no_replica_servers.size());
    *any_replica_server = replica_servers.front();
    *no_replica_server = no_replica_servers.front();
  }
};

// Slow tablet replica is not evicted while bootstrapping
// as long as there are no data modifications in the consensus.
TEST_F(SlowTabletBootstrapTest, TestSlowBootstrap) {
  SKIP_IF_SLOW_NOT_ALLOWED();

  string any_replica_server, no_replica_server;
  NO_FATALS(SetUpCluster(&any_replica_server, &no_replica_server));

  // Shutdown any tablet server with tablet's replica.
  auto ts = cluster_->tablet_server_by_uuid(any_replica_server);
  ASSERT_NE(nullptr, ts);

  ts->mutable_flags()->emplace_back("--tablet_bootstrap_inject_latency_ms=4000");
  ts->Shutdown();
  ASSERT_OK(ts->Restart());
  SleepFor(MonoDelta::FromSeconds(7));

  ASSERT_EVENTUALLY([&]() {
    ValidateConsensusStateChanged(any_replica_server, no_replica_server, kTimeout);
  });
}

// If replica restarts after many data modifications,
// it falls behind and is removed from consensus.
TEST_F(SlowTabletBootstrapTest, TestFallBehind) {
  SKIP_IF_SLOW_NOT_ALLOWED();

  string any_replica_server, no_replica_server;
  NO_FATALS(SetUpCluster(&any_replica_server, &no_replica_server));

  // Shutdown any tablet server with tablet's replica,
  // add data, then restart and cause it to fall behind.
  NO_FATALS(CauseSpecificFollowerToFallBehindLogGC(tablet_servers_,
      any_replica_server, nullptr, nullptr,
      BehindWalGcBehavior::SHUTDOWN_RESTART));

  ASSERT_EVENTUALLY([&]() {
    ValidateConsensusStateChanged(no_replica_server, any_replica_server, kTimeout);
  });
}

// If there many data modifications during slow replica bootstrap,
// it falls behind and is removed from consensus.
TEST_F(SlowTabletBootstrapTest, TestFallBehindSlowBootstrap) {
  SKIP_IF_SLOW_NOT_ALLOWED();

  string any_replica_server, no_replica_server;
  NO_FATALS(SetUpCluster(&any_replica_server, &no_replica_server));

  // Shutdown any tablet server with tablet replica.
  auto ts = cluster_->tablet_server_by_uuid(any_replica_server);
  ASSERT_NE(nullptr, ts);

  // Inject delay into next tablet replica bootstrap.
  // When replica finish bootstrapping, it will find that it was left behind
  // and was removed from consensus.
  ts->mutable_flags()->emplace_back("--tablet_bootstrap_inject_latency_ms=3000");

  TServerDetails* replica = FindOrDie(tablet_servers_, any_replica_server);
  ts->Shutdown();
  ASSERT_OK(ts->Restart());

  // Find a leader.
  TServerDetails* leader = nullptr;
  ASSERT_OK(GetLeaderReplicaWithRetries(tablet_id_, &leader));
  ASSERT_NE(leader, nullptr);
  ASSERT_NE(leader, replica);
  int leader_index = cluster_->tablet_server_index_by_uuid(leader->uuid());

  TestWorkload workload(cluster_.get());
  workload.set_table_name(kTableId);
  workload.set_timeout_allowed(true);
  workload.set_payload_bytes(128 * 1024); // Write ops of size 128KB.
  workload.set_write_batch_size(1);
  workload.set_num_write_threads(4);
  workload.Setup();
  workload.Start();

  LOG(INFO) << "Waiting until we've written at least 4MB...";
  while (workload.rows_inserted() < 8 * 4) {
    SleepFor(MonoDelta::FromMilliseconds(10));
  }
  workload.StopAndJoin();

  LOG(INFO) << "Waiting for log GC on " << leader->uuid();
  // Some WAL segments must exist, but wal segment 1 must not exist.
  ASSERT_OK(inspect_->WaitForFilePatternInTabletWalDirOnTs(
      leader_index, tablet_id_, { "wal-" }, { "wal-000000001" }));
  LOG(INFO) << "Log GC complete on " << leader->uuid();

  ASSERT_EVENTUALLY([&]() {
    ValidateConsensusStateChanged(no_replica_server, any_replica_server, kTimeout);
  });
}

} // namespace kudu