YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

// Copyright (c) YugaByte, Inc.

//

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

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

//

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

//

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

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

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

// under the License.

//

#include <algorithm>

#include <map>

#include <string>

#include <utility>

#include <chrono>

#include <boost/assign.hpp>

#include <gflags/gflags.h>

#include <gtest/gtest.h>

#include "yb/common/ql_value.h"

#include "yb/common/schema.h"

#include "yb/common/wire_protocol.h"

#include "yb/cdc/cdc_service.h"

#include "yb/cdc/cdc_service.pb.h"

#include "yb/cdc/cdc_service.proxy.h"

#include "yb/client/client.h"

#include "yb/client/client-test-util.h"

#include "yb/client/meta_cache.h"

#include "yb/client/schema.h"

#include "yb/client/session.h"

#include "yb/client/table.h"

#include "yb/client/table_alterer.h"

#include "yb/client/table_creator.h"

#include "yb/client/table_handle.h"

#include "yb/client/transaction.h"

#include "yb/client/yb_op.h"

#include "yb/gutil/stl_util.h"

#include "yb/gutil/strings/join.h"

#include "yb/gutil/strings/substitute.h"

#include "yb/integration-tests/cdc_test_util.h"

#include "yb/integration-tests/mini_cluster.h"

#include "yb/integration-tests/twodc_test_base.h"

#include "yb/integration-tests/yb_mini_cluster_test_base.h"

#include "yb/master/catalog_manager_if.h"

#include "yb/master/master_defaults.h"

#include "yb/master/mini_master.h"

#include "yb/master/master_replication.proxy.h"

#include "yb/master/master-test-util.h"

#include "yb/master/cdc_consumer_registry_service.h"

#include "yb/rpc/rpc_controller.h"

#include "yb/server/hybrid_clock.h"

#include "yb/tablet/tablet.h"

#include "yb/tablet/tablet_peer.h"

#include "yb/tserver/mini_tablet_server.h"

#include "yb/tserver/tablet_server.h"

#include "yb/tserver/ts_tablet_manager.h"

#include "yb/tserver/cdc_consumer.h"

#include "yb/util/atomic.h"

#include "yb/util/faststring.h"

#include "yb/util/metrics.h"

#include "yb/util/random.h"

#include "yb/util/status_log.h"

#include "yb/util/stopwatch.h"

#include "yb/util/test_util.h"

using namespace std::literals;

DECLARE_int32(replication_factor);

DECLARE_bool(enable_ysql);

DECLARE_bool(TEST_twodc_write_hybrid_time);

DECLARE_int32(cdc_wal_retention_time_secs);

DECLARE_int32(replication_failure_delay_exponent);

DECLARE_double(TEST_respond_write_failed_probability);

DECLARE_int32(cdc_max_apply_batch_num_records);

DECLARE_int32(async_replication_idle_delay_ms);

DECLARE_int32(async_replication_max_idle_wait);

DECLARE_int32(external_intent_cleanup_secs);

DECLARE_int32(yb_num_shards_per_tserver);

DECLARE_uint64(TEST_yb_inbound_big_calls_parse_delay_ms);

DECLARE_int64(rpc_throttle_threshold_bytes);

DECLARE_bool(enable_automatic_tablet_splitting);

namespace yb {

using client::YBClient;

using client::YBClientBuilder;

using client::YBColumnSchema;

using client::YBError;

using client::YBSchema;

using client::YBSchemaBuilder;

using client::YBSession;

using client::YBTable;

using client::YBTableAlterer;

using client::YBTableCreator;

using client::YBTableType;

using client::YBTableName;

using master::MiniMaster;

using tserver::MiniTabletServer;

using tserver::enterprise::CDCConsumer;

namespace enterprise {

using SessionTransactionPair = std::pair<client::YBSessionPtr, client::YBTransactionPtr>;

class TwoDCTest : public TwoDCTestBase, public testing::WithParamInterface<TwoDCTestParams> {

 public:

  Result<std::vector<std::shared_ptr<client::YBTable>>> SetUpWithParams(

      const std::vector<uint32_t>& num_consumer_tablets,

      const std::vector<uint32_t>& num_producer_tablets,

      uint32_t replication_factor,

      uint32_t num_masters = 1) {

    FLAGS_enable_ysql = false;

    TwoDCTestBase::SetUp();

    FLAGS_cdc_max_apply_batch_num_records = GetParam().batch_size;

    FLAGS_cdc_enable_replicate_intents = GetParam().enable_replicate_intents;

    FLAGS_yb_num_shards_per_tserver = 1;

    MiniClusterOptions opts;

    opts.num_tablet_servers = replication_factor;

    opts.num_masters = num_masters;

    FLAGS_replication_factor = replication_factor;

    opts.cluster_id = "producer";

    producer_cluster_.mini_cluster_ = std::make_unique<MiniCluster>(opts);

    RETURN_NOT_OK(producer_cluster()->StartSync());

    RETURN_NOT_OK(producer_cluster()->WaitForTabletServerCount(replication_factor));

    opts.cluster_id = "consumer";

    consumer_cluster_.mini_cluster_ = std::make_unique<MiniCluster>(opts);

    RETURN_NOT_OK(consumer_cluster()->StartSync());

    RETURN_NOT_OK(consumer_cluster()->WaitForTabletServerCount(replication_factor));

    producer_cluster_.client_ = VERIFY_RESULT(producer_cluster()->CreateClient());

    consumer_cluster_.client_ = VERIFY_RESULT(consumer_cluster()->CreateClient());

    RETURN_NOT_OK(clock_->Init());

    producer_cluster_.txn_mgr_.emplace(producer_client(), clock_, client::LocalTabletFilter());

    consumer_cluster_.txn_mgr_.emplace(consumer_client(), clock_, client::LocalTabletFilter());

    YBSchemaBuilder b;

    b.AddColumn("c0")->Type(INT32)->NotNull()->HashPrimaryKey();

    // Create transactional table.

    TableProperties table_properties;

    table_properties.SetTransactional(true);

    b.SetTableProperties(table_properties);

    CHECK_OK(b.Build(&schema_));

    YBSchema consumer_schema;

    table_properties.SetDefaultTimeToLive(0);

    b.SetTableProperties(table_properties);

    CHECK_OK(b.Build(&consumer_schema));

    if (num_consumer_tablets.size() != num_producer_tablets.size()) {

      return STATUS(IllegalState,

                    Format("Num consumer tables: $0 num producer tables: $1 must be equal.",

                           num_consumer_tablets.size(), num_producer_tablets.size()));

    }

    std::vector<YBTableName> tables;

    std::vector<std::shared_ptr<client::YBTable>> yb_tables;

    for (uint32_t i = 0; i < num_consumer_tablets.size(); i++) {

      RETURN_NOT_OK(CreateTable(i, num_producer_tablets[i], producer_client(), &tables));

      std::shared_ptr<client::YBTable> producer_table;

      RETURN_NOT_OK(producer_client()->OpenTable(tables[i * 2], &producer_table));

      yb_tables.push_back(producer_table);

      RETURN_NOT_OK(CreateTable(i, num_consumer_tablets[i], consumer_client(),

                                consumer_schema, &tables));

      std::shared_ptr<client::YBTable> consumer_table;

      RETURN_NOT_OK(consumer_client()->OpenTable(tables[(i * 2) + 1], &consumer_table));

      yb_tables.push_back(consumer_table);

    }

    return yb_tables;

  }

  Result<YBTableName> CreateTable(YBClient* client, const std::string& namespace_name,

                                  const std::string& table_name, uint32_t num_tablets,

                                  const YBSchema* schema = nullptr) {

    YBTableName table(YQL_DATABASE_CQL, namespace_name, table_name);

    RETURN_NOT_OK(client->CreateNamespaceIfNotExists(table.namespace_name(),

                                                     table.namespace_type()));

    if (!schema) {

      schema = &schema_;

    }

    // Add a table, make sure it reports itself.

    std::unique_ptr<YBTableCreator> table_creator(client->NewTableCreator());

        RETURN_NOT_OK(table_creator->table_name(table)

                          .schema(schema)

                          .table_type(YBTableType::YQL_TABLE_TYPE)

                          .num_tablets(num_tablets)

                          .Create());

    return table;

  }

  Status CreateTable(

      uint32_t idx, uint32_t num_tablets, YBClient* client, std::vector<YBTableName>* tables) {

    auto table = VERIFY_RESULT(CreateTable(client, kNamespaceName, Format("test_table_$0", idx),

                                           num_tablets));

    tables->push_back(table);

    return Status::OK();

  }

  Status CreateTable(uint32_t idx, uint32_t num_tablets, YBClient* client, YBSchema schema,

                     std::vector<YBTableName>* tables) {

    auto table = VERIFY_RESULT(CreateTable(client, kNamespaceName, Format("test_table_$0", idx),

                                           num_tablets, &schema));

    tables->push_back(table);

    return Status::OK();

  }

  void WriteWorkload(uint32_t start, uint32_t end, YBClient* client, const YBTableName& table,

                     bool delete_op = false) {

    auto session = client->NewSession();

    client::TableHandle table_handle;

    ASSERT_OK(table_handle.Open(table, client));

    std::vector<std::shared_ptr<client::YBqlOp>> ops;

    LOG(INFO) << "Writing " << end-start << (delete_op ? " deletes" : " inserts");

    for (uint32_t i = start; i < end; i++) {

      auto op = delete_op ? table_handle.NewDeleteOp() : table_handle.NewInsertOp();

      int32_t key = i;

      auto req = op->mutable_request();

      QLAddInt32HashValue(req, key);

      ASSERT_OK(session->ApplyAndFlush(op));

    }

  }

  void DeleteWorkload(uint32_t start, uint32_t end, YBClient* client, const YBTableName& table) {

    WriteWorkload(start, end, client, table, true /* delete_op */);

  }

  std::vector<string> ScanToStrings(const YBTableName& table_name, YBClient* client) {

    client::TableHandle table;

    EXPECT_OK(table.Open(table_name, client));

    auto result = ScanTableToStrings(table);

    std::sort(result.begin(), result.end());

    return result;

  }

  Status VerifyWrittenRecords(const YBTableName& producer_table,

                              const YBTableName& consumer_table,

                              int timeout_secs = kRpcTimeout) {

    return LoggedWaitFor([=]() -> Result<bool> {

      auto producer_results = ScanToStrings(producer_table, producer_client());

      auto consumer_results = ScanToStrings(consumer_table, consumer_client());

      return producer_results == consumer_results;

    }, MonoDelta::FromSeconds(timeout_secs), "Verify written records");

  }

  Status VerifyNumRecords(const YBTableName& table, YBClient* client, size_t expected_size) {

    return LoggedWaitFor([=]() -> Result<bool> {

      auto results = ScanToStrings(table, client);

      return results.size() == expected_size;

    }, MonoDelta::FromSeconds(kRpcTimeout), "Verify number of records");

  }

  Result<SessionTransactionPair> CreateSessionWithTransaction(

      YBClient* client, client::TransactionManager* txn_mgr) {

    auto session = client->NewSession();

    auto transaction = std::make_shared<client::YBTransaction>(txn_mgr);

    ReadHybridTime read_time;

    RETURN_NOT_OK(transaction->Init(IsolationLevel::SNAPSHOT_ISOLATION, read_time));

    session->SetTransaction(transaction);

    return std::make_pair(session, transaction);

  }

  void WriteIntents(uint32_t start, uint32_t end, YBClient* client,

                    const std::shared_ptr<YBSession>& session, const YBTableName& table,

                    bool delete_op = false) {

    client::TableHandle table_handle;

    ASSERT_OK(table_handle.Open(table, client));

    std::vector<std::shared_ptr<client::YBqlOp>> ops;

    for (uint32_t i = start; i < end; i++) {

      auto op = delete_op ? table_handle.NewDeleteOp() : table_handle.NewInsertOp();

      int32_t key = i;

      auto req = op->mutable_request();

      QLAddInt32HashValue(req, key);

      ASSERT_OK(session->ApplyAndFlush(op));

    }

  }

  void WriteTransactionalWorkload(uint32_t start, uint32_t end, YBClient* client,

                                  client::TransactionManager* txn_mgr, const YBTableName& table,

                                  bool delete_op = false) {

    auto pair = ASSERT_RESULT(CreateSessionWithTransaction(client, txn_mgr));

    ASSERT_NO_FATALS(WriteIntents(start, end, client, pair.first, table, delete_op));

    ASSERT_OK(pair.second->CommitFuture().get());

  }

 private:

  server::ClockPtr clock_{new server::HybridClock()};

  YBSchema schema_;

};

INSTANTIATE_TEST_CASE_P(TwoDCTestParams, TwoDCTest,

                        ::testing::Values(TwoDCTestParams(1, true), TwoDCTestParams(1, false),

                                          TwoDCTestParams(0, true), TwoDCTestParams(0, false)));

TEST_P(TwoDCTest, SetupUniverseReplication) {

  auto tables = ASSERT_RESULT(SetUpWithParams({8, 4}, {6, 6}, 3));

  std::vector<std::shared_ptr<client::YBTable>> producer_tables;

  // tables contains both producer and consumer universe tables (alternately).

  // Pick out just the producer tables from the list.

  producer_tables.reserve(tables.size() / 2);

  for (size_t i = 0; i < tables.size(); i += 2) {

    producer_tables.push_back(tables[i]);

  }

  ASSERT_OK(SetupUniverseReplication(

      producer_cluster(), consumer_cluster(), consumer_client(), kUniverseId, producer_tables));

  // Verify that universe was setup on consumer.

  master::GetUniverseReplicationResponsePB resp;

  ASSERT_OK(VerifyUniverseReplication(consumer_cluster(), consumer_client(), kUniverseId, &resp));

  ASSERT_EQ(resp.entry().producer_id(), kUniverseId);

  ASSERT_EQ(resp.entry().tables_size(), producer_tables.size());

  for (uint32_t i = 0; i < producer_tables.size(); i++) {

    ASSERT_EQ(resp.entry().tables(i), producer_tables[i]->id());

  }

  // Verify that CDC streams were created on producer for all tables.

  for (size_t i = 0; i < producer_tables.size(); i++) {

    master::ListCDCStreamsResponsePB stream_resp;

    ASSERT_OK(GetCDCStreamForTable(producer_tables[i]->id(), &stream_resp));

    ASSERT_EQ(stream_resp.streams_size(), 1);

    ASSERT_EQ(stream_resp.streams(0).table_id().Get(0), producer_tables[i]->id());

  }

  ASSERT_OK(DeleteUniverseReplication(kUniverseId));

}

TEST_P(TwoDCTest, SetupUniverseReplicationErrorChecking) {

  auto tables = ASSERT_RESULT(SetUpWithParams({1}, {1}, 1));

  rpc::RpcController rpc;

  auto master_proxy = std::make_shared<master::MasterReplicationProxy>(

      &consumer_client()->proxy_cache(),

      ASSERT_RESULT(consumer_cluster()->GetLeaderMiniMaster())->bound_rpc_addr());

  {

    rpc.Reset();

    rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));

    master::SetupUniverseReplicationRequestPB setup_universe_req;

    master::SetupUniverseReplicationResponsePB setup_universe_resp;

    ASSERT_OK(master_proxy->SetupUniverseReplication(

      setup_universe_req, &setup_universe_resp, &rpc));

    ASSERT_TRUE(setup_universe_resp.has_error());

    std::string prefix = "Producer universe ID must be provided";

    ASSERT_TRUE(setup_universe_resp.error().status().message().substr(0, prefix.size()) == prefix);

  }

  {

    rpc.Reset();

    rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));

    master::SetupUniverseReplicationRequestPB setup_universe_req;

    setup_universe_req.set_producer_id(kUniverseId);

    master::SetupUniverseReplicationResponsePB setup_universe_resp;

    ASSERT_OK(master_proxy->SetupUniverseReplication(

      setup_universe_req, &setup_universe_resp, &rpc));

    ASSERT_TRUE(setup_universe_resp.has_error());

    std::string prefix = "Producer master address must be provided";

    ASSERT_TRUE(setup_universe_resp.error().status().message().substr(0, prefix.size()) == prefix);

  }

  {

    rpc.Reset();

    rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));

    master::SetupUniverseReplicationRequestPB setup_universe_req;

    setup_universe_req.set_producer_id(kUniverseId);

    string master_addr = producer_cluster()->GetMasterAddresses();

    auto hp_vec = ASSERT_RESULT(HostPort::ParseStrings(master_addr, 0));

    HostPortsToPBs(hp_vec, setup_universe_req.mutable_producer_master_addresses());

    setup_universe_req.add_producer_table_ids("a");

    setup_universe_req.add_producer_table_ids("b");

    setup_universe_req.add_producer_bootstrap_ids("c");

    master::SetupUniverseReplicationResponsePB setup_universe_resp;

    ASSERT_OK(master_proxy->SetupUniverseReplication(

      setup_universe_req, &setup_universe_resp, &rpc));

    ASSERT_TRUE(setup_universe_resp.has_error());

    std::string prefix = "Number of bootstrap ids must be equal to number of tables";

    ASSERT_TRUE(setup_universe_resp.error().status().message().substr(0, prefix.size()) == prefix);

  }

  {

    rpc.Reset();

    rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));

    master::SetupUniverseReplicationRequestPB setup_universe_req;

    master::SetupUniverseReplicationResponsePB setup_universe_resp;

    setup_universe_req.set_producer_id(kUniverseId);

    string master_addr = consumer_cluster()->GetMasterAddresses();

    auto hp_vec = ASSERT_RESULT(HostPort::ParseStrings(master_addr, 0));

    HostPortsToPBs(hp_vec, setup_universe_req.mutable_producer_master_addresses());

    setup_universe_req.add_producer_table_ids("prod_table_id_1");

    setup_universe_req.add_producer_table_ids("prod_table_id_2");

    setup_universe_req.add_producer_bootstrap_ids("prod_bootstrap_id_1");

    setup_universe_req.add_producer_bootstrap_ids("prod_bootstrap_id_2");

    ASSERT_OK(master_proxy->SetupUniverseReplication(

      setup_universe_req, &setup_universe_resp, &rpc));

    ASSERT_TRUE(setup_universe_resp.has_error());

    std::string prefix = "Duplicate between request master addresses";

    ASSERT_TRUE(setup_universe_resp.error().status().message().substr(0, prefix.size()) == prefix);

  }

  {

    rpc.Reset();

    rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));

    master::SetupUniverseReplicationRequestPB setup_universe_req;

    master::SetupUniverseReplicationResponsePB setup_universe_resp;

    master::SysClusterConfigEntryPB cluster_info;

    auto& cm = ASSERT_RESULT(consumer_cluster()->GetLeaderMiniMaster())->catalog_manager();

    CHECK_OK(cm.GetClusterConfig(&cluster_info));

    setup_universe_req.set_producer_id(cluster_info.cluster_uuid());

    string master_addr = producer_cluster()->GetMasterAddresses();

    auto hp_vec = ASSERT_RESULT(HostPort::ParseStrings(master_addr, 0));

    HostPortsToPBs(hp_vec, setup_universe_req.mutable_producer_master_addresses());

    setup_universe_req.add_producer_table_ids("prod_table_id_1");

    setup_universe_req.add_producer_table_ids("prod_table_id_2");

    setup_universe_req.add_producer_bootstrap_ids("prod_bootstrap_id_1");

    setup_universe_req.add_producer_bootstrap_ids("prod_bootstrap_id_2");

    ASSERT_OK(master_proxy->SetupUniverseReplication(

      setup_universe_req, &setup_universe_resp, &rpc));

    ASSERT_TRUE(setup_universe_resp.has_error());

    std::string prefix = "The request UUID and cluster UUID are identical.";

    ASSERT_TRUE(setup_universe_resp.error().status().message().substr(0, prefix.size()) == prefix);

  }

}

TEST_P(TwoDCTest, SetupUniverseReplicationWithProducerBootstrapId) {

  constexpr int kNTabletsPerTable = 1;

  std::vector<uint32_t> tables_vector = {kNTabletsPerTable, kNTabletsPerTable};

  auto tables = ASSERT_RESULT(SetUpWithParams(tables_vector, tables_vector, 3));

  std::unique_ptr<client::YBClient> client;

  std::unique_ptr<cdc::CDCServiceProxy> producer_cdc_proxy;

  client = ASSERT_RESULT(consumer_cluster()->CreateClient());

  producer_cdc_proxy = std::make_unique<cdc::CDCServiceProxy>(

      &client->proxy_cache(),

      HostPort::FromBoundEndpoint(producer_cluster()->mini_tablet_server(0)->bound_rpc_addr()));

  // tables contains both producer and consumer universe tables (alternately).

  // Pick out just the producer tables from the list.

  std::vector<std::shared_ptr<client::YBTable>> producer_tables;

  std::vector<std::shared_ptr<client::YBTable>> consumer_tables;

  producer_tables.reserve(tables.size() / 2);

  consumer_tables.reserve(tables.size() / 2);

  for (size_t i = 0; i < tables.size(); i ++) {

    if (i % 2 == 0) {

      producer_tables.push_back(tables[i]);

    } else {

      consumer_tables.push_back(tables[i]);

    }

  }

  // 1. Write some data so that we can verify that only new records get replicated

  for (const auto& producer_table : producer_tables) {

    LOG(INFO) << "Writing records for table " << producer_table->name().ToString();

    WriteWorkload(0, 100, producer_client(), producer_table->name());

  }

  SleepFor(MonoDelta::FromSeconds(10));

  cdc::BootstrapProducerRequestPB req;

  cdc::BootstrapProducerResponsePB resp;

  for (const auto& producer_table : producer_tables) {

    req.add_table_ids(producer_table->id());

  }

  rpc::RpcController rpc;

  ASSERT_OK(producer_cdc_proxy->BootstrapProducer(req, &resp, &rpc));

  ASSERT_FALSE(resp.has_error());

  ASSERT_EQ(resp.cdc_bootstrap_ids().size(), producer_tables.size());

  int table_idx = 0;

  for (const auto& bootstrap_id : resp.cdc_bootstrap_ids()) {

    LOG(INFO) << "Got bootstrap id " << bootstrap_id

              << " for table " << producer_tables[table_idx++]->name().table_name();

  }

  std::unordered_map<std::string, int> tablet_bootstraps;

  // Verify that for each of the table's tablets, a new row in cdc_state table with the returned

  // id was inserted.

  client::TableHandle table;

  client::YBTableName cdc_state_table(

      YQL_DATABASE_CQL, master::kSystemNamespaceName, master::kCdcStateTableName);

  ASSERT_OK(table.Open(cdc_state_table, producer_client()));

  // 2 tables with 8 tablets each.

  ASSERT_EQ(tables_vector.size() * kNTabletsPerTable, boost::size(client::TableRange(table)));

  int nrows = 0;

  for (const auto& row : client::TableRange(table)) {

    nrows++;

    string stream_id = row.column(0).string_value();

    tablet_bootstraps[stream_id]++;

    string checkpoint = row.column(2).string_value();

    auto s = OpId::FromString(checkpoint);

    ASSERT_OK(s);

    OpId op_id = *s;

    ASSERT_GT(op_id.index, 0);

    LOG(INFO) << "Bootstrap id " << stream_id

              << " for tablet " << row.column(1).string_value();

  }

  ASSERT_EQ(tablet_bootstraps.size(), producer_tables.size());

  // Check that each bootstrap id has 8 tablets.

  for (const auto& e : tablet_bootstraps) {

    ASSERT_EQ(e.second, kNTabletsPerTable);

  }

  // Map table -> bootstrap_id. We will need when setting up replication.

  std::unordered_map<TableId, std::string> table_bootstrap_ids;

  for (int i = 0; i < resp.cdc_bootstrap_ids_size(); i++) {

    table_bootstrap_ids[req.table_ids(i)] = resp.cdc_bootstrap_ids(i);

  }

  // 2. Setup replication.

  master::SetupUniverseReplicationRequestPB setup_universe_req;

  master::SetupUniverseReplicationResponsePB setup_universe_resp;

  setup_universe_req.set_producer_id(kUniverseId);

  string master_addr = producer_cluster()->GetMasterAddresses();

  auto hp_vec = ASSERT_RESULT(HostPort::ParseStrings(master_addr, 0));

  HostPortsToPBs(hp_vec, setup_universe_req.mutable_producer_master_addresses());

  setup_universe_req.mutable_producer_table_ids()->Reserve(

      narrow_cast<int>(producer_tables.size()));

  for (const auto& producer_table : producer_tables) {

    setup_universe_req.add_producer_table_ids(producer_table->id());

    const auto& iter = table_bootstrap_ids.find(producer_table->id());

    ASSERT_NE(iter, table_bootstrap_ids.end());

    setup_universe_req.add_producer_bootstrap_ids(iter->second);

  }

  auto master_proxy = std::make_shared<master::MasterReplicationProxy>(

      &consumer_client()->proxy_cache(),

      ASSERT_RESULT(consumer_cluster()->GetLeaderMiniMaster())->bound_rpc_addr());

  rpc.Reset();

  rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));

  ASSERT_OK(master_proxy->SetupUniverseReplication(

    setup_universe_req, &setup_universe_resp, &rpc));

  ASSERT_FALSE(setup_universe_resp.has_error());

  // 3. Verify everything is setup correctly.

  master::GetUniverseReplicationResponsePB get_universe_replication_resp;

  ASSERT_OK(VerifyUniverseReplication(consumer_cluster(), consumer_client(), kUniverseId,

      &get_universe_replication_resp));

  ASSERT_OK(CorrectlyPollingAllTablets(

      consumer_cluster(), narrow_cast<int32_t>(tables_vector.size() * kNTabletsPerTable)));

  // 4. Write more data.

  for (const auto& producer_table : producer_tables) {

    WriteWorkload(1000, 1005, producer_client(), producer_table->name());

  }

  // 5. Verify that only new writes get replicated to consumer since we bootstrapped the producer

  // after we had already written some data, therefore the old data (whatever was there before we

  // bootstrapped the producer) should not be replicated.

  auto data_replicated_correctly = [&]() {

    for (const auto& consumer_table : consumer_tables) {

      LOG(INFO) << "Checking records for table " << consumer_table->name().ToString();

      std::vector<std::string> expected_results;

      for (int key = 1000; key < 1005; key++) {

        expected_results.emplace_back("{ int32:" + std::to_string(key) + " }");

      }

      std::sort(expected_results.begin(), expected_results.end());

      auto consumer_results = ScanToStrings(consumer_table->name(), consumer_client());

      std::sort(consumer_results.begin(), consumer_results.end());

      if (expected_results.size() != consumer_results.size()) {

        return false;

      }

      for (size_t idx = 0; idx < expected_results.size(); idx++) {

        if (expected_results[idx] != consumer_results[idx]) {

          return false;

        }

      }

    }

    return true;

  };

  ASSERT_OK(WaitFor([&]() { return data_replicated_correctly(); },

                    MonoDelta::FromSeconds(20), "IsDataReplicatedCorrectly"));

}

// Test for #2250 to verify that replication for tables with the same prefix gets set up correctly.

TEST_P(TwoDCTest, SetupUniverseReplicationMultipleTables) {

  // Setup the two clusters without any tables.

  auto tables = ASSERT_RESULT(SetUpWithParams({}, {}, 1));

  // Create tables with the same prefix.

  std::string table_names[2] = {"table", "table_index"};

  std::vector<std::shared_ptr<client::YBTable>> producer_tables;

  for (int i = 0; i < 2; i++) {

    auto t = ASSERT_RESULT(CreateTable(producer_client(), kNamespaceName, table_names[i], 3));

    std::shared_ptr<client::YBTable> producer_table;

    ASSERT_OK(producer_client()->OpenTable(t, &producer_table));

    producer_tables.push_back(producer_table);

  }

  for (int i = 0; i < 2; i++) {

    ASSERT_RESULT(CreateTable(consumer_client(), kNamespaceName, table_names[i], 3));

  }

  // Setup universe replication on both these tables.

  ASSERT_OK(SetupUniverseReplication(

      producer_cluster(), consumer_cluster(), consumer_client(), kUniverseId, producer_tables));

  // Verify that universe was setup on consumer.

  master::GetUniverseReplicationResponsePB resp;

  ASSERT_OK(VerifyUniverseReplication(consumer_cluster(), consumer_client(), kUniverseId, &resp));

  ASSERT_EQ(resp.entry().producer_id(), kUniverseId);

  ASSERT_EQ(resp.entry().tables_size(), producer_tables.size());

  for (uint32_t i = 0; i < producer_tables.size(); i++) {

    ASSERT_EQ(resp.entry().tables(i), producer_tables[i]->id());

  }

  ASSERT_OK(DeleteUniverseReplication(kUniverseId));

}

TEST_P(TwoDCTest, SetupUniverseReplicationLargeTableCount) {

  if (IsSanitizer()) {

    LOG(INFO) << "Skipping slow test";

    return;

  }

  // Setup the two clusters without any tables.

  auto tables = ASSERT_RESULT(SetUpWithParams({}, {}, 1));

  FLAGS_enable_automatic_tablet_splitting = false;

  // Create a large number of tables to test the performance of setup_replication.

  int table_count = 2;

  int amplification[2] = {1, 5};

  MonoDelta setup_latency[2];

  std::string table_prefix = "stress_table_";

  bool passed_test = false;

  for (int retries = 0; retries < 3 && !passed_test; ++retries) {

    for (int a : {0, 1}) {

      std::vector<std::shared_ptr<client::YBTable>> producer_tables;

      for (int i = 0; i < table_count * amplification[a]; i++) {

        std::string cur_table =

            table_prefix + std::to_string(amplification[a]) + "-" + std::to_string(i);

        ASSERT_RESULT(CreateTable(consumer_client(), kNamespaceName, cur_table, 3));

        auto t = ASSERT_RESULT(CreateTable(producer_client(), kNamespaceName, cur_table, 3));

        std::shared_ptr<client::YBTable> producer_table;

        ASSERT_OK(producer_client()->OpenTable(t, &producer_table));

        producer_tables.push_back(producer_table);

      }

      // Add delays to all rpc calls to simulate live environment and ensure the test is IO bound.

      FLAGS_TEST_yb_inbound_big_calls_parse_delay_ms = 200;

      FLAGS_rpc_throttle_threshold_bytes = 200;

      auto start_time = CoarseMonoClock::Now();

      // Setup universe replication on all tables.

      ASSERT_OK(SetupUniverseReplication(

          producer_cluster(), consumer_cluster(), consumer_client(), kUniverseId, producer_tables));

      // Verify that universe was setup on consumer.

      master::GetUniverseReplicationResponsePB resp;

      ASSERT_OK(

          VerifyUniverseReplication(consumer_cluster(), consumer_client(), kUniverseId, &resp));

      ASSERT_EQ(resp.entry().producer_id(), kUniverseId);

      ASSERT_EQ(resp.entry().tables_size(), producer_tables.size());

      for (uint32_t i = 0; i < producer_tables.size(); i++) {

        ASSERT_EQ(resp.entry().tables(i), producer_tables[i]->id());

      }

      setup_latency[a] = CoarseMonoClock::Now() - start_time;

      LOG(INFO) << "SetupReplication [" << a << "] took: " << setup_latency[a].ToSeconds() << "s";

      // Remove delays for cleanup and next setup.

      FLAGS_TEST_yb_inbound_big_calls_parse_delay_ms = 0;

      ASSERT_OK(DeleteUniverseReplication(kUniverseId));

    }

    // We increased our table count by 5x, but we shouldn't have a linear latency increase.

    passed_test = (setup_latency[1] < setup_latency[0] * 3);

  }

  ASSERT_TRUE(passed_test);

}

TEST_P(TwoDCTest, BootstrapAndSetupLargeTableCount) {

  if (IsSanitizer()) {

    LOG(INFO) << "Skipping slow test";

    return;

  }

  // Setup the two clusters without any tables.

  auto tables = ASSERT_RESULT(SetUpWithParams({}, {}, 1));

  FLAGS_enable_automatic_tablet_splitting = false;

  // Create a medium, then large number of tables to test the performance of our CLI commands.

  int table_count = 2;

  int amplification[2] = {1, 5};

  MonoDelta bootstrap_latency[2];

  MonoDelta setup_latency[2];

  std::string table_prefix = "stress_table_";

  bool passed_test = false;

  for (int retries = 0; retries < 3 && !passed_test; ++retries) {

    for (int a : {0, 1}) {

      std::vector<std::shared_ptr<client::YBTable>> producer_tables;

      for (int i = 0; i < table_count * amplification[a]; i++) {

        std::string cur_table =

            table_prefix + std::to_string(amplification[a]) + "-" + std::to_string(i);

        ASSERT_RESULT(CreateTable(consumer_client(), kNamespaceName, cur_table, 3));

        auto t = ASSERT_RESULT(CreateTable(producer_client(), kNamespaceName, cur_table, 3));

        std::shared_ptr<client::YBTable> producer_table;

        ASSERT_OK(producer_client()->OpenTable(t, &producer_table));

        producer_tables.push_back(producer_table);

      }

      // Add delays to all rpc calls to simulate live environment and ensure the test is IO bound.

      FLAGS_TEST_yb_inbound_big_calls_parse_delay_ms = 200;

      FLAGS_rpc_throttle_threshold_bytes = 200;

      // Performance test of BootstrapProducer.

      cdc::BootstrapProducerResponsePB boot_resp;

      {

        cdc::BootstrapProducerRequestPB req;

        for (const auto& producer_table : producer_tables) {

          req.add_table_ids(producer_table->id());

        }

        auto start_time = CoarseMonoClock::Now();

        auto producer_cdc_proxy = std::make_unique<cdc::CDCServiceProxy>(

            &producer_client()->proxy_cache(),

            HostPort::FromBoundEndpoint(

                producer_cluster()->mini_tablet_server(0)->bound_rpc_addr()));

        rpc::RpcController rpc;

        ASSERT_OK(producer_cdc_proxy->BootstrapProducer(req, &boot_resp, &rpc));

        ASSERT_FALSE(boot_resp.has_error());

        ASSERT_EQ(boot_resp.cdc_bootstrap_ids().size(), producer_tables.size());

        bootstrap_latency[a] = CoarseMonoClock::Now() - start_time;

        LOG(INFO) << "BootstrapProducer [" << a << "] took: " << bootstrap_latency[a].ToSeconds()

                  << "s";

      }

      // Performance test of SetupReplication, with Bootstrap IDs.

      {

        auto start_time = CoarseMonoClock::Now();

        // Calling the SetupUniverse API directly so we can use producer_bootstrap_ids.

        master::SetupUniverseReplicationRequestPB req;

        master::SetupUniverseReplicationResponsePB resp;

        req.set_producer_id(kUniverseId);

        auto master_addrs = producer_cluster()->GetMasterAddresses();

        auto vec = ASSERT_RESULT(HostPort::ParseStrings(master_addrs, 0));

        HostPortsToPBs(vec, req.mutable_producer_master_addresses());

        for (const auto& table : producer_tables) {

          req.add_producer_table_ids(table->id());

        }

        for (const auto& bootstrap_id : boot_resp.cdc_bootstrap_ids()) {

          req.add_producer_bootstrap_ids(bootstrap_id);

        }

        auto master_proxy = std::make_shared<master::MasterReplicationProxy>(

            &consumer_client()->proxy_cache(),

            ASSERT_RESULT(consumer_cluster()->GetLeaderMiniMaster())->bound_rpc_addr());

        ASSERT_OK(WaitFor(

            [&]() -> Result<bool> {

              rpc::RpcController rpc;

              rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));

              if (!master_proxy->SetupUniverseReplication(req, &resp, &rpc).ok()) {

                return false;

              }

              if (resp.has_error()) {

                return false;

              }

              return true;

            },

            MonoDelta::FromSeconds(30), "Setup universe replication"));

        // Verify that universe was setup on consumer.

        {

          master::GetUniverseReplicationResponsePB resp;

          ASSERT_OK(

              VerifyUniverseReplication(consumer_cluster(), consumer_client(), kUniverseId, &resp));

          ASSERT_EQ(resp.entry().producer_id(), kUniverseId);

          ASSERT_EQ(resp.entry().tables_size(), producer_tables.size());

          for (uint32_t i = 0; i < producer_tables.size(); i++) {

            ASSERT_EQ(resp.entry().tables(i), producer_tables[i]->id());

          }

        }

        setup_latency[a] = CoarseMonoClock::Now() - start_time;

        LOG(INFO) << "SetupReplication [" << a << "] took: " << setup_latency[a].ToSeconds() << "s";

      }

      // Remove delays for cleanup and next setup.

      FLAGS_TEST_yb_inbound_big_calls_parse_delay_ms = 0;

      ASSERT_OK(DeleteUniverseReplication(kUniverseId));

    }

    // We increased our table count by 5x, but we shouldn't have a linear latency increase.

    // ASSERT_LT(bootstrap_latency[1], bootstrap_latency[0] * 5);

    passed_test = (setup_latency[1] < setup_latency[0] * 3);

  }

  ASSERT_TRUE(passed_test);

}

TEST_P(TwoDCTest, PollWithConsumerRestart) {

  // Avoid long delays with node failures so we can run with more aggressive test timing

  FLAGS_replication_failure_delay_exponent = 7; // 2^7 == 128ms

  uint32_t replication_factor = NonTsanVsTsan(3, 1);

  auto tables = ASSERT_RESULT(SetUpWithParams({4}, {4}, replication_factor));

  ASSERT_OK(SetupUniverseReplication(

      producer_cluster(), consumer_cluster(), consumer_client(), kUniverseId,

      {tables[0]} /* all producer tables */));

  // After creating the cluster, make sure all tablets being polled for.

  ASSERT_OK(CorrectlyPollingAllTablets(consumer_cluster(), 4));

  consumer_cluster()->mini_tablet_server(0)->Shutdown();

  // After shutting down a single consumer node, the other consumers should pick up the slack.

  if (replication_factor > 1) {

    ASSERT_OK(CorrectlyPollingAllTablets(consumer_cluster(), 4));

  }

  ASSERT_OK(consumer_cluster()->mini_tablet_server(0)->Start());

  // After restarting the node.

  ASSERT_OK(CorrectlyPollingAllTablets(consumer_cluster(), 4));

  ASSERT_OK(consumer_cluster()->RestartSync());

  // After consumer cluster restart.

  ASSERT_OK(CorrectlyPollingAllTablets(consumer_cluster(), 4));

  ASSERT_OK(DeleteUniverseReplication(kUniverseId));

}

TEST_P(TwoDCTest, PollWithProducerNodesRestart) {

  // Avoid long delays with node failures so we can run with more aggressive test timing

  FLAGS_replication_failure_delay_exponent = 7; // 2^7 == 128ms

  uint32_t replication_factor = 3, tablet_count = 4, master_count = 3;

  auto tables = ASSERT_RESULT(

      SetUpWithParams({tablet_count}, {tablet_count}, replication_factor,  master_count));

  ASSERT_OK(SetupUniverseReplication(

    producer_cluster(), consumer_cluster(), consumer_client(), kUniverseId,

    {tables[0]} /* all producer tables */, false /* leader_only */));

  // After creating the cluster, make sure all tablets being polled for.

  ASSERT_OK(CorrectlyPollingAllTablets(consumer_cluster(), 4));

  // Stop the Master and wait for failover.

  LOG(INFO) << "Failover to new Master";

  MiniMaster* old_master = ASSERT_RESULT(producer_cluster()->GetLeaderMiniMaster());

  ASSERT_OK(old_master->WaitUntilCatalogManagerIsLeaderAndReadyForTests());

  ASSERT_RESULT(producer_cluster()->GetLeaderMiniMaster())->Shutdown();

  MiniMaster* new_master = ASSERT_RESULT(producer_cluster()->GetLeaderMiniMaster());

  ASSERT_NE(nullptr, new_master);

  ASSERT_NE(old_master, new_master);

  ASSERT_OK(producer_cluster()->WaitForAllTabletServers());

  // Stop a TServer on the Producer after failing its master.

  producer_cluster()->mini_tablet_server(0)->Shutdown();

  // This Verifies: