text: // 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 <chrono>
  #include <utility>
  #include <boost/assign.hpp>
  #include <gtest/gtest.h>
  
  #include "yb/cdc/cdc_service.h"
  #include "yb/cdc/cdc_service.pb.h"
  
  #include "yb/client/client-test-util.h"
  #include "yb/client/client.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/tablet/transaction_participant.h"
  #include "yb/client/yb_op.h"
  
  #include "yb/common/common.pb.h"
  #include "yb/common/entity_ids.h"
  #include "yb/common/ql_value.h"
  #include "yb/common/wire_protocol.h"
  
  #include "yb/gutil/stl_util.h"
  #include "yb/gutil/strings/join.h"
  #include "yb/gutil/strings/substitute.h"
  
  #include "yb/integration-tests/cdcsdk_test_base.h"
  #include "yb/integration-tests/mini_cluster.h"
  
  #include "yb/master/master.h"
  #include "yb/master/master_client.pb.h"
  #include "yb/master/master_cluster.pb.h"
  #include "yb/master/master_cluster.proxy.h"
  #include "yb/master/master_ddl.pb.h"
  #include "yb/master/master_replication.proxy.h"
  #include "yb/master/mini_master.h"
  
  #include "yb/rpc/rpc_controller.h"
  
  #include "yb/tablet/tablet.h"
  #include "yb/tablet/tablet_peer.h"
  
  #include "yb/tserver/cdc_consumer.h"
  #include "yb/tserver/mini_tablet_server.h"
  #include "yb/tserver/tablet_server.h"
  #include "yb/tserver/ts_tablet_manager.h"
  #include "yb/tserver/tserver_admin.proxy.h"
  
  #include "yb/util/monotime.h"
  #include "yb/util/random_util.h"
  #include "yb/util/result.h"
  #include "yb/util/test_macros.h"
  #include "yb/util/thread.h"
  
  #include "yb/yql/pgwrapper/libpq_utils.h"
  #include "yb/yql/pgwrapper/pg_wrapper.h"
  
  DECLARE_int64(cdc_intent_retention_ms);
  DECLARE_bool(enable_update_local_peer_min_index);
  DECLARE_int32(update_min_cdc_indices_interval_secs);
  DECLARE_bool(stream_truncate_record);
  DECLARE_int32(cdc_state_checkpoint_update_interval_ms);
  
  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::YBTableName;
  using client::YBTableType;
  using master::GetNamespaceInfoResponsePB;
  using master::MiniMaster;
  using tserver::MiniTabletServer;
  using tserver::enterprise::CDCConsumer;
  
  using pgwrapper::GetInt32;
  using pgwrapper::PGConn;
  using pgwrapper::PGResultPtr;
  using pgwrapper::PgSupervisor;
  using pgwrapper::ToString;
  
  using rpc::RpcController;
  
  namespace cdc {
  namespace enterprise {
  
  class CDCSDKYsqlTest : public CDCSDKTestBase {
   public:
    struct ExpectedRecord {
      int32_t key;
      int32_t value;
    };
  
    Result<string> GetUniverseId(Cluster* cluster) {
      yb::master::GetMasterClusterConfigRequestPB req;
      yb::master::GetMasterClusterConfigResponsePB resp;
  
      master::MasterClusterProxy master_proxy(
          &cluster->client_->proxy_cache(),
          VERIFY_RESULT(cluster->mini_cluster_->GetLeaderMasterBoundRpcAddr()));
  
      rpc::RpcController rpc;
      rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));
      RETURN_NOT_OK(master_proxy.GetMasterClusterConfig(req, &resp, &rpc));
      if (resp.has_error()) {
        return STATUS(IllegalState, "Error getting cluster config");
      }
      return resp.cluster_config().cluster_uuid();
    }
  
    Status DropDB(Cluster* cluster) {
      const std::string db_name = "testdatabase";
      RETURN_NOT_OK(CreateDatabase(&test_cluster_, db_name, true));
      auto conn = VERIFY_RESULT(cluster->ConnectToDB(db_name));
      RETURN_NOT_OK(conn.ExecuteFormat("DROP DATABASE $0", kNamespaceName));
      return Status::OK();
    }
  
    Status TruncateTable(Cluster* cluster, const std::vector<string>& table_ids) {
      RETURN_NOT_OK(cluster->client_->TruncateTables(table_ids));
      return Status::OK();
    }
  
    // The range is exclusive of end i.e. [start, end)
    Status WriteRows(uint32_t start, uint32_t end, Cluster* cluster) {
      auto conn = VERIFY_RESULT(cluster->ConnectToDB(kNamespaceName));
      LOG(INFO) << "Writing " << end - start << " row(s)";
  
      for (uint32_t i = start; i < end; ++i) {
        RETURN_NOT_OK(conn.ExecuteFormat(
            "INSERT INTO $0($1, $2) VALUES ($3, $4)", kTableName, kKeyColumnName, kValueColumnName, i,
            i + 1));
      }
      return Status::OK();
    }
  
    Status WriteRowsHelper(uint32_t start, uint32_t end, Cluster* cluster, bool flag) {
      auto conn = VERIFY_RESULT(cluster->ConnectToDB(kNamespaceName));
      LOG(INFO) << "Writing " << end - start << " row(s) within transaction";
  
      RETURN_NOT_OK(conn.Execute("BEGIN"));
      for (uint32_t i = start; i < end; ++i) {
        RETURN_NOT_OK(conn.ExecuteFormat(
            "INSERT INTO $0($1, $2) VALUES ($3, $4)", kTableName, kKeyColumnName, kValueColumnName, i,
            i + 1));
      }
      if (flag) {
        RETURN_NOT_OK(conn.Execute("COMMIT"));
      } else {
        RETURN_NOT_OK(conn.Execute("ABORT"));
      }
      return Status::OK();
    }
  
    Status UpdateRows(uint32_t key, uint32_t value, Cluster* cluster) {
      auto conn = VERIFY_RESULT(cluster->ConnectToDB(kNamespaceName));
      LOG(INFO) << "Updating row for key " << key << " with value " << value;
      RETURN_NOT_OK(conn.ExecuteFormat(
          "UPDATE $0 SET $1 = $2 WHERE $3 = $4", kTableName, kValueColumnName, value, kKeyColumnName,
          key));
      return Status::OK();
    }
  
    Status DeleteRows(uint32_t key, Cluster* cluster) {
      auto conn = VERIFY_RESULT(cluster->ConnectToDB(kNamespaceName));
      LOG(INFO) << "Deleting row for key " << key;
      RETURN_NOT_OK(
          conn.ExecuteFormat("DELETE FROM $0 WHERE $1 = $2", kTableName, kKeyColumnName, key));
      return Status::OK();
    }
  
    Result<google::protobuf::RepeatedPtrField<master::TabletLocationsPB>> SetUpCluster() {
      RETURN_NOT_OK(SetUpWithParams(3, 1, false));
      auto table = EXPECT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
      google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
      RETURN_NOT_OK(test_client()->GetTablets(table, 0, &tablets, nullptr));
      return tablets;
    }
  
    Result<GetChangesResponsePB> UpdateCheckpoint(
        const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets,
        GetChangesResponsePB* change_resp) {
      GetChangesRequestPB change_req2;
      GetChangesResponsePB change_resp2;
      PrepareChangeRequest(
          &change_req2, stream_id, tablets, change_resp->cdc_sdk_checkpoint().index(),
          change_resp->cdc_sdk_checkpoint().term(), change_resp->cdc_sdk_checkpoint().key(),
          change_resp->cdc_sdk_checkpoint().write_id(),
          change_resp->cdc_sdk_checkpoint().snapshot_time());
      RpcController get_changes_rpc;
      RETURN_NOT_OK(cdc_proxy_->GetChanges(change_req2, &change_resp2, &get_changes_rpc));
      if (change_resp2.has_error()) {
        return StatusFromPB(change_resp2.error().status());
      }
  
      return change_resp2;
    }
  
    std::unique_ptr<tserver::TabletServerAdminServiceProxy> GetTServerAdminProxy(
        const uint32_t tserver_index) {
      auto tserver = test_cluster()->mini_tablet_server(tserver_index);
      return std::make_unique<tserver::TabletServerAdminServiceProxy>(
          &tserver->server()->proxy_cache(), HostPort::FromBoundEndpoint(tserver->bound_rpc_addr()));
    }
  
    Status GetIntentCounts(const uint32_t tserver_index, int64* num_intents) {
      tserver::CountIntentsRequestPB req;
      tserver::CountIntentsResponsePB resp;
      RpcController rpc;
  
      auto ts_admin_service_proxy = GetTServerAdminProxy(tserver_index);
      rpc.set_timeout(MonoDelta::FromSeconds(kRpcTimeout));
      RETURN_NOT_OK(ts_admin_service_proxy->CountIntents(req, &resp, &rpc));
      *num_intents = resp.num_intents();
      return Status::OK();
    }
  
    void PrepareChangeRequest(
        GetChangesRequestPB* change_req, const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets, int64 index = 0,
        int64 term = 0, std::string key = "", int32_t write_id = 0, int64 snapshot_time = 0) {
      change_req->set_stream_id(stream_id);
      change_req->set_tablet_id(tablets.Get(0).tablet_id());
      change_req->mutable_from_cdc_sdk_checkpoint()->set_index(index);
      change_req->mutable_from_cdc_sdk_checkpoint()->set_term(term);
      change_req->mutable_from_cdc_sdk_checkpoint()->set_key(key);
      change_req->mutable_from_cdc_sdk_checkpoint()->set_write_id(write_id);
      change_req->mutable_from_cdc_sdk_checkpoint()->set_snapshot_time(snapshot_time);
    }
  
    void PrepareChangeRequest(
        GetChangesRequestPB* change_req, const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets,
        const CDCSDKCheckpointPB& cp) {
      change_req->set_stream_id(stream_id);
      change_req->set_tablet_id(tablets.Get(0).tablet_id());
      change_req->mutable_from_cdc_sdk_checkpoint()->set_term(cp.term());
      change_req->mutable_from_cdc_sdk_checkpoint()->set_index(cp.index());
      change_req->mutable_from_cdc_sdk_checkpoint()->set_key(cp.key());
      change_req->mutable_from_cdc_sdk_checkpoint()->set_write_id(cp.write_id());
    }
  
    void PrepareSetCheckpointRequest(
        SetCDCCheckpointRequestPB* set_checkpoint_req,
        const CDCStreamId stream_id,
        google::protobuf::RepeatedPtrField<master::TabletLocationsPB>
            tablets,
        const OpId& op_id,
        bool initial_checkpoint) {
      set_checkpoint_req->set_stream_id(stream_id);
      set_checkpoint_req->set_initial_checkpoint(initial_checkpoint);
      set_checkpoint_req->set_tablet_id(tablets.Get(0).tablet_id());
      set_checkpoint_req->mutable_checkpoint()->mutable_op_id()->set_term(op_id.term);
      set_checkpoint_req->mutable_checkpoint()->mutable_op_id()->set_index(op_id.index);
    }
  
    Result<SetCDCCheckpointResponsePB> SetCDCCheckpoint(
        const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets,
        const OpId& op_id = OpId::Min(),
        bool initial_checkpoint = true) {
      RpcController set_checkpoint_rpc;
      SetCDCCheckpointRequestPB set_checkpoint_req;
      SetCDCCheckpointResponsePB set_checkpoint_resp;
      auto deadline = CoarseMonoClock::now() + test_client()->default_rpc_timeout();
      set_checkpoint_rpc.set_deadline(deadline);
      PrepareSetCheckpointRequest(&set_checkpoint_req, stream_id, tablets, op_id, initial_checkpoint);
      Status st =
          cdc_proxy_->SetCDCCheckpoint(set_checkpoint_req, &set_checkpoint_resp, &set_checkpoint_rpc);
  
      RETURN_NOT_OK(st);
      return set_checkpoint_resp;
    }
  
    Result<std::vector<OpId>> GetCDCCheckpoint(
        const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets) {
      RpcController get_checkpoint_rpc;
      GetCheckpointRequestPB get_checkpoint_req;
      GetCheckpointResponsePB get_checkpoint_resp;
      auto deadline = CoarseMonoClock::now() + test_client()->default_rpc_timeout();
      get_checkpoint_rpc.set_deadline(deadline);
  
      std::vector<OpId> op_ids;
      for (auto tablet : tablets) {
        get_checkpoint_req.set_stream_id(stream_id);
        get_checkpoint_req.set_tablet_id(tablets.Get(0).tablet_id());
        RETURN_NOT_OK(
            cdc_proxy_->GetCheckpoint(get_checkpoint_req, &get_checkpoint_resp, &get_checkpoint_rpc));
        op_ids.push_back(OpId::FromPB(get_checkpoint_resp.checkpoint().op_id()));
      }
      return op_ids;
    }
  
    void AssertKeyValue(const CDCSDKProtoRecordPB& record, const int32_t& key, const int32_t& value) {
      ASSERT_EQ(key, record.row_message().new_tuple(0).datum_int32());
      ASSERT_EQ(value, record.row_message().new_tuple(1).datum_int32());
    }
  
    void CheckRecord(
        const CDCSDKProtoRecordPB& record, CDCSDKYsqlTest::ExpectedRecord expected_records,
        uint32_t* count) {
      // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
      switch (record.row_message().op()) {
        case RowMessage::DDL: {
          ASSERT_EQ(record.row_message().table(), kTableName);
          count[0]++;
        } break;
        case RowMessage::INSERT: {
          AssertKeyValue(record, expected_records.key, expected_records.value);
          ASSERT_EQ(record.row_message().table(), kTableName);
          count[1]++;
        } break;
        case RowMessage::UPDATE: {
          AssertKeyValue(record, expected_records.key, expected_records.value);
          ASSERT_EQ(record.row_message().table(), kTableName);
          count[2]++;
        } break;
        case RowMessage::DELETE: {
          ASSERT_EQ(record.row_message().old_tuple(0).datum_int32(), expected_records.key);
          ASSERT_EQ(record.row_message().table(), kTableName);
          count[3]++;
        } break;
        case RowMessage::READ: {
          AssertKeyValue(record, expected_records.key, expected_records.value);
          ASSERT_EQ(record.row_message().table(), kTableName);
          count[4]++;
        } break;
        case RowMessage::TRUNCATE: {
          count[5]++;
        } break;
        default:
          ASSERT_FALSE(true);
          break;
      }
    }
  
    void CheckCount(const uint32_t* expected_count, uint32_t* count) {
      for (int i = 0; i < 6; i++) {
        ASSERT_EQ(expected_count[i], count[i]);
      }
    }
  
    Result<GetChangesResponsePB> GetChangesFromCDC(
        const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets,
        const CDCSDKCheckpointPB* cp = nullptr) {
      GetChangesRequestPB change_req;
      GetChangesResponsePB change_resp;
  
      if (cp == nullptr) {
        PrepareChangeRequest(&change_req, stream_id, tablets);
      } else {
        PrepareChangeRequest(&change_req, stream_id, tablets, *cp);
      }
  
      RpcController get_changes_rpc;
      RETURN_NOT_OK(cdc_proxy_->GetChanges(change_req, &change_resp, &get_changes_rpc));
  
      if (change_resp.has_error()) {
        return StatusFromPB(change_resp.error().status());
      }
  
      return change_resp;
    }
  
    Result<GetChangesResponsePB> GetChangesFromCDCSnapshot(
        const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets) {
      GetChangesRequestPB change_req;
      GetChangesResponsePB change_resp;
      PrepareChangeRequest(&change_req, stream_id, tablets, -1, -1, "", -1, 0);
      RpcController get_changes_rpc;
      RETURN_NOT_OK(cdc_proxy_->GetChanges(change_req, &change_resp, &get_changes_rpc));
  
      if (change_resp.has_error()) {
        return StatusFromPB(change_resp.error().status());
      }
      return change_resp;
    }
  
    void TestGetChanges(
        const uint32_t replication_factor, bool add_tables_without_primary_key = false) {
      ASSERT_OK(SetUpWithParams(replication_factor, 1, false));
  
      auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
  
      if (add_tables_without_primary_key) {
        // Adding tables without primary keys, they should not disturb any CDC related processes.
        std::string tables_wo_pk[] = {"table_wo_pk_1", "table_wo_pk_2", "table_wo_pk_3"};
        for (const auto& table_name : tables_wo_pk) {
          auto temp = ASSERT_RESULT(
              CreateTable(&test_cluster_, kNamespaceName, table_name, 1 /* num_tablets */, false));
        }
      }
  
      google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
      ASSERT_OK(test_client()->GetTablets(
          table, 0, &tablets,
          /* partition_list_version =*/nullptr));
      ASSERT_EQ(tablets.size(), 1);
  
      std::string table_id = ASSERT_RESULT(GetTableId(&test_cluster_, kNamespaceName, kTableName));
      CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
  
      auto resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
      ASSERT_FALSE(resp.has_error());
      ASSERT_OK(WriteRows(0 /* start */, 1 /* end */, &test_cluster_));
  
      const uint32_t expected_records_size = 1;
      int expected_record[] = {0 /* key */, 1 /* value */};
  
      SleepFor(MonoDelta::FromSeconds(5));
      GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
      uint32_t record_size = change_resp.cdc_sdk_proto_records_size();
      uint32_t ins_count = 0;
      for (uint32_t i = 0; i < record_size; ++i) {
        if (change_resp.cdc_sdk_proto_records(i).row_message().op() == RowMessage::INSERT) {
          const CDCSDKProtoRecordPB record = change_resp.cdc_sdk_proto_records(i);
          AssertKeyValue(record, expected_record[0], expected_record[1]);
          ++ins_count;
        }
      }
      LOG(INFO) << "Got " << ins_count << " insert records";
      ASSERT_EQ(expected_records_size, ins_count);
    }
  
    void TestIntentGarbageCollectionFlag(
        const uint32_t num_tservers,
        const bool set_flag_to_a_smaller_value,
        const uint32_t cdc_intent_retention_ms) {
      if (set_flag_to_a_smaller_value) {
        FLAGS_cdc_intent_retention_ms = cdc_intent_retention_ms;
      }
      FLAGS_enable_update_local_peer_min_index = false;
      FLAGS_update_min_cdc_indices_interval_secs = 1;
  
      ASSERT_OK(SetUpWithParams(num_tservers, 1, false));
  
      auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
  
      google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
      ASSERT_OK(
          test_client()->GetTablets(table, 0, &tablets, /* partition_list_version = */ nullptr));
  
      TabletId table_id = ASSERT_RESULT(GetTableId(&test_cluster_, kNamespaceName, kTableName));
      CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream(CDCCheckpointType::IMPLICIT));
      auto resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
      ASSERT_FALSE(resp.has_error());
  
      // Call GetChanges once to set the initial value in the cdc_state table.
      GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
      // This will write one row with PK = 0.
      ASSERT_OK(WriteRows(0 /* start */, 1 /* end */, &test_cluster_));
  
      // Count intents here, they should be 0 here.
      for (uint32_t i = 0; i < num_tservers; ++i) {
        int64 intents_count = 0;
        ASSERT_OK(GetIntentCounts(i, &intents_count));
        ASSERT_EQ(0, intents_count);
      }
  
      ASSERT_OK(WriteRowsHelper(1, 2, &test_cluster_, true));
      ASSERT_OK(test_client()->FlushTables(
          {table.table_id()}, /* add_indexes = */ false,
          /* timeout_secs = */ 30, /* is_compaction = */ false));
  
      // Sleep for 60s for the background thread to update the consumer op_id so that garbage
      // collection can happen.
      vector<int64> intent_counts(num_tservers, -1);
      ASSERT_OK(WaitFor(
          [this, &num_tservers, &set_flag_to_a_smaller_value, &intent_counts, &stream_id,
           &tablets]() -> Result<bool> {
            uint32_t i = 0;
            while (i < num_tservers) {
              // Call GetChanges once to set the initial value in the cdc_state table.
              auto result = GetChangesFromCDC(stream_id, tablets);
              if (!result.ok()) {
                return false;
              }
              yb::cdc::GetChangesResponsePB change_resp = *result;
              if (change_resp.has_error()) {
                return false;
              }
  
              auto status = GetIntentCounts(i, &intent_counts[i]);
              if (!status.ok()) {
                continue;
              }
  
              if (set_flag_to_a_smaller_value) {
                if (intent_counts[i] != 0) {
                  continue;
                }
              }
              i++;
            }
            return true;
          },
          MonoDelta::FromSeconds(60), "Waiting for all the tservers intent counts"));
  
      for (uint32_t i = 0; i < num_tservers; ++i) {
        if (set_flag_to_a_smaller_value) {
          ASSERT_EQ(0, intent_counts[i]);
        } else {
          ASSERT_NE(0, intent_counts[i]);
        }
      }
    }
  
    void TestSetCDCCheckpoint(const uint32_t num_tservers, bool initial_checkpoint) {
      ASSERT_OK(SetUpWithParams(num_tservers, 1, false));
      auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
      google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
      ASSERT_OK(
          test_client()->GetTablets(table, 0, &tablets, /* partition_list_version = */ nullptr));
  
      TabletId table_id = ASSERT_RESULT(GetTableId(&test_cluster_, kNamespaceName, kTableName));
      CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream(CDCCheckpointType::IMPLICIT));
      auto resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
      ASSERT_FALSE(resp.has_error());
      auto checkpoints = ASSERT_RESULT(GetCDCCheckpoint(stream_id, tablets));
      for (auto op_id : checkpoints) {
        ASSERT_EQ(OpId(0, 0), op_id);
      }
  
      resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets, OpId(1, 3)));
      ASSERT_FALSE(resp.has_error());
  
      checkpoints = ASSERT_RESULT(GetCDCCheckpoint(stream_id, tablets));
  
      for (auto op_id : checkpoints) {
        ASSERT_EQ(OpId(1, 3), op_id);
      }
  
      resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets, OpId(1, -3)));
      ASSERT_TRUE(resp.has_error());
  
      resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets, OpId(-2, 1)));
      ASSERT_TRUE(resp.has_error());
    }
  
    Result<GetChangesResponsePB> VerifyIfDDLRecordPresent(
        const CDCStreamId& stream_id,
        const google::protobuf::RepeatedPtrField<master::TabletLocationsPB>& tablets,
        bool expect_ddl_record, bool is_first_call, const CDCSDKCheckpointPB* cp = nullptr) {
      GetChangesRequestPB req;
      GetChangesResponsePB resp;
  
      if (cp == nullptr) {
        PrepareChangeRequest(&req, stream_id, tablets);
      } else {
        PrepareChangeRequest(&req, stream_id, tablets, *cp);
      }
  
      // The default value for need_schema_info is false.
      if (expect_ddl_record) {
        req.set_need_schema_info(true);
      }
  
      RpcController get_changes_rpc;
      RETURN_NOT_OK(cdc_proxy_->GetChanges(req, &resp, &get_changes_rpc));
  
      if (resp.has_error()) {
        return StatusFromPB(resp.error().status());
      }
  
      auto record = resp.cdc_sdk_proto_records(0);
  
      // If it's the first call to GetChanges, we will get a DDL record irrespective of the
      // value of need_schema_info.
      if (is_first_call || expect_ddl_record) {
        EXPECT_EQ(record.row_message().op(), RowMessage::DDL);
      } else {
        EXPECT_NE(record.row_message().op(), RowMessage::DDL);
      }
  
      return resp;
    }
  };
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestBaseFunctions)) {
    // setting up a cluster with 3 RF
    ASSERT_OK(SetUpWithParams(3, 1, false /* colocated */));
  
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
  
    ASSERT_FALSE(table.is_cql_namespace());
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestLoadInsertionOnly)) {
    // set up an RF3 cluster
    ASSERT_OK(SetUpWithParams(3, 1, false));
  
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
  
    ASSERT_OK(WriteRows(0, 10, &test_cluster_));
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(GetChangesWithRF1)) {
    TestGetChanges(1 /* replication factor */);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(GetChangesWithRF3)) {
    TestGetChanges(3 /* replication factor */);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(GetChanges_TablesWithNoPKPresentInDB)) {
    TestGetChanges(3 /* replication_factor */, true /* add_tables_without_primary_key */);
  }
  
  // Insert a single row.
  // Expected records: (DDL, INSERT).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(SingleShardInsertWithAutoCommit)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRows(1 /* start */, 2 /* end */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 1, 0, 0, 0, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}, {1, 2}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    uint32_t record_size = change_resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[1] << " insert record";
    CheckCount(expected_count, count);
  }
  
  // Begin transaction, perform some operations and abort transaction.
  // Expected records: 1 (DDL).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(AbortAllWriteOperations)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
    ASSERT_OK(WriteRowsHelper(1 /* start */, 4 /* end */, &test_cluster_, false));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 0, 0, 0, 0, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    uint32_t record_size = change_resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[1] << " insert record and " << count[0] << " ddl record";
    CheckCount(expected_count, count);
  }
  
  // Insert one row, update the inserted row.
  // Expected records: (DDL, INSERT, UPDATE).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(SingleShardUpdateWithAutoCommit)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRows(1 /* start */, 2 /* end */, &test_cluster_));
    ASSERT_OK(UpdateRows(1 /* key */, 1 /* value */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 1, 1, 0, 0, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}, {1, 2}, {1, 1}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    uint32_t record_size = change_resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[1] << " insert record and " << count[2] << " update record";
    CheckCount(expected_count, count);
  }
  
  // Insert 3 rows, update 2 of them.
  // Expected records: (DDL, 3 INSERT, 2 UPDATE).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(SingleShardUpdateRows)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRows(1 /* start */, 4 /* end */, &test_cluster_));
    ASSERT_OK(UpdateRows(1 /* key */, 1 /* value */, &test_cluster_));
    ASSERT_OK(UpdateRows(2 /* key */, 2 /* value */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 3, 2, 0, 0, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}, {1, 2}, {2, 3}, {3, 4}, {1, 1}, {2, 2}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    uint32_t record_size = change_resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[1] << " insert record and " << count[2] << " update record";
    CheckCount(expected_count, count);
  }
  
  // Insert one row, delete inserted row.
  // Expected records: (DDL, INSERT, DELETE).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(SingleShardDeleteWithAutoCommit)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRows(1 /* start */, 2 /* end */, &test_cluster_));
    ASSERT_OK(DeleteRows(1 /* key */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 1, 0, 1, 0, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}, {1, 2}, {1, 0}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    uint32_t record_size = change_resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[1] << " insert record and " << count[3] << " delete record";
    CheckCount(expected_count, count);
  }
  
  // Insert 4 rows.
  // Expected records: (DDL, INSERT, INSERT, INSERT, INSERT).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(SingleShardInsert4Rows)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRows(1 /* start */, 5 /* end */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 4, 0, 0, 0, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}, {1, 2}, {2, 3}, {3, 4}, {4, 5}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    uint32_t record_size = change_resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[1] << " insert records";
    CheckCount(expected_count, count);
  }
  
  // Insert a row before snapshot. Insert a row after snapshot.
  // Expected records: (DDL, READ) and (DDL, INSERT).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(InsertBeforeAfterSnapshot)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRows(1 /* start */, 2 /* end */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {2, 1, 0, 0, 1, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records_before_snapshot[] = {{0, 0}, {1, 2}};
    ExpectedRecord expected_records_after_snapshot[] = {{0, 0}, {2, 3}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDCSnapshot(stream_id, tablets));
    GetChangesResponsePB change_resp_updated =
        ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
  
    uint32_t record_size = change_resp_updated.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp_updated.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records_before_snapshot[i], count);
    }
  
    ASSERT_OK(WriteRows(2 /* start */, 3 /* end */, &test_cluster_));
    GetChangesResponsePB change_resp_after_snapshot =
        ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp_updated));
    uint32_t record_size_after_snapshot = change_resp_after_snapshot.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size_after_snapshot; ++i) {
      const CDCSDKProtoRecordPB record = change_resp_after_snapshot.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records_after_snapshot[i], count);
    }
    CheckCount(expected_count, count);
  }
  
  // Begin transaction, insert one row, commit transaction, enable snapshot
  // Expected records: (DDL, READ).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(InsertSingleRowSnapshot)) {
    ASSERT_OK(SetUpWithParams(3, 1, false));
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
    google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
    ASSERT_OK(test_client()->GetTablets(table, 0, &tablets, nullptr));
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRowsHelper(1 /* start */, 2 /* end */, &test_cluster_, true));
    ASSERT_OK(test_client()->FlushTables(
        {table.table_id()}, /* add_indexes = */ false,
        /* timeout_secs = */ 30, /* is_compaction = */ false));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 0, 0, 0, 1, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}, {1, 2}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDCSnapshot(stream_id, tablets));
    GetChangesResponsePB change_resp_updated =
        ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
  
    uint32_t record_size = change_resp_updated.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp_updated.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[4] << " read record and " << count[0] << " ddl record";
    CheckCount(expected_count, count);
  }
  
  // Begin transaction, insert one row, commit transaction, update, enable snapshot
  // Expected records: (DDL, READ).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(UpdateInsertedRowSnapshot)) {
    ASSERT_OK(SetUpWithParams(3, 1, false));
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
    google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
    ASSERT_OK(test_client()->GetTablets(table, 0, &tablets, nullptr));
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRowsHelper(1 /* start */, 2 /* end */, &test_cluster_, true));
    ASSERT_OK(test_client()->FlushTables(
        {table.table_id()}, /* add_indexes = */ false,
        /* timeout_secs = */ 30, /* is_compaction = */ false));
    ASSERT_OK(UpdateRows(1 /* key */, 1 /* value */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 0, 0, 0, 1, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}, {1, 1}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDCSnapshot(stream_id, tablets));
    GetChangesResponsePB change_resp_updated =
        ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
  
    uint32_t record_size = change_resp_updated.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp_updated.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[4] << " read record and " << count[0] << " ddl record";
    CheckCount(expected_count, count);
  }
  
  // Begin transaction, insert one row, commit transaction, delete, enable snapshot
  // Expected records: (DDL).
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(DeleteInsertedRowSnapshot)) {
    ASSERT_OK(SetUpWithParams(3, 1, false));
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
    google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
    ASSERT_OK(test_client()->GetTablets(table, 0, &tablets, nullptr));
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    ASSERT_OK(WriteRowsHelper(1 /* start */, 2 /* end */, &test_cluster_, true));
    ASSERT_OK(test_client()->FlushTables(
        {table.table_id()}, /* add_indexes = */ false,
        /* timeout_secs = */ 30, /* is_compaction = */ false));
    ASSERT_OK(DeleteRows(1 /* key */, &test_cluster_));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count[] = {1, 0, 0, 0, 0, 0};
    uint32_t count[] = {0, 0, 0, 0, 0, 0};
  
    ExpectedRecord expected_records[] = {{0, 0}};
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDCSnapshot(stream_id, tablets));
    GetChangesResponsePB change_resp_updated =
        ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
  
    uint32_t record_size = change_resp_updated.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = change_resp_updated.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records[i], count);
    }
    LOG(INFO) << "Got " << count[4] << " read record and " << count[0] << " ddl record";
    CheckCount(expected_count, count);
  }
  
  // Insert 10K rows using a thread and after a while enable snapshot.
  // Expected sum of READs and INSERTs is 10K.
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(InsertBeforeDuringSnapshot)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    // 10K records inserted using a thread.
    std::vector<std::thread> threads;
    threads.emplace_back(
        [&]() { ASSERT_OK(WriteRows(1 /* start */, 10001 /* end */, &test_cluster_)); });
    SleepFor(MonoDelta::FromMilliseconds(100));
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDCSnapshot(stream_id, tablets));
  
    // Count the number of snapshot READs.
    uint32_t reads_snapshot = 0;
    bool end_snapshot = false;
    while (true) {
      GetChangesResponsePB change_resp_updated =
          ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
      uint32_t record_size = change_resp_updated.cdc_sdk_proto_records_size();
      uint32_t read_count = 0;
      for (uint32_t i = 0; i < record_size; ++i) {
        const CDCSDKProtoRecordPB record = change_resp_updated.cdc_sdk_proto_records(i);
        if (record.row_message().op() == RowMessage::READ) {
          read_count++;
        } else if (record.row_message().op() == RowMessage::INSERT) {
          end_snapshot = true;
          break;
        }
      }
      if (end_snapshot) {
        break;
      }
      reads_snapshot += read_count;
      change_resp = change_resp_updated;
      if (reads_snapshot == 10000) {
        break;
      }
    }
  
    for (auto& t : threads) {
      t.join();
    }
  
    LOG(INFO) << "Insertion of records using threads has completed.";
  
    // Count the number of INSERTS.
    uint32_t inserts_snapshot = 0;
    while (true) {
      GetChangesResponsePB change_resp_after_snapshot =
          ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
      uint32_t record_size_after_snapshot = change_resp_after_snapshot.cdc_sdk_proto_records_size();
      if (record_size_after_snapshot == 0) {
        break;
      }
      uint32_t insert_count = 0;
      for (uint32_t i = 0; i < record_size_after_snapshot; ++i) {
        const CDCSDKProtoRecordPB record = change_resp_after_snapshot.cdc_sdk_proto_records(i);
        if (record.row_message().op() == RowMessage::INSERT) {
          insert_count++;
        }
      }
      inserts_snapshot += insert_count;
      change_resp = change_resp_after_snapshot;
    }
    LOG(INFO) << "Got " << reads_snapshot + inserts_snapshot << " total (read + insert) record";
    ASSERT_EQ(reads_snapshot + inserts_snapshot, 10000);
  }
  
  // Insert 10K rows using a thread and after a while enable snapshot.
  // After snapshot completes, insert 10K rows using threads.
  // Expected sum of READs and INSERTs is 20K.
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(InsertBeforeDuringAfterSnapshot)) {
    auto tablets = ASSERT_RESULT(SetUpCluster());
    ASSERT_EQ(tablets.size(), 1);
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
  
    // 10K records inserted using a thread.
    std::vector<std::thread> threads;
    threads.emplace_back(
        [&]() { ASSERT_OK(WriteRows(1 /* start */, 10001 /* end */, &test_cluster_)); });
    SleepFor(MonoDelta::FromMilliseconds(100));
  
    GetChangesResponsePB change_resp = ASSERT_RESULT(GetChangesFromCDCSnapshot(stream_id, tablets));
  
    // Count the number of snapshot READs.
    uint32_t reads_snapshot = 0;
    bool end_snapshot = false;
    while (true) {
      GetChangesResponsePB change_resp_updated =
          ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
      uint32_t record_size = change_resp_updated.cdc_sdk_proto_records_size();
      uint32_t read_count = 0;
      for (uint32_t i = 0; i < record_size; ++i) {
        const CDCSDKProtoRecordPB record = change_resp_updated.cdc_sdk_proto_records(i);
        if (record.row_message().op() == RowMessage::READ) {
          read_count++;
        } else if (record.row_message().op() == RowMessage::INSERT) {
          end_snapshot = true;
          break;
        }
      }
      if (end_snapshot) {
        break;
      }
      reads_snapshot += read_count;
      change_resp = change_resp_updated;
      if (reads_snapshot == 10000) {
        break;
      }
    }
  
    // Two threads used to insert records after the snapshot is over.
    threads.emplace_back(
        [&]() { ASSERT_OK(WriteRows(10001 /* start */, 15001 /* end */, &test_cluster_)); });
    threads.emplace_back(
        [&]() { ASSERT_OK(WriteRows(15001 /* start */, 20001 /* end */, &test_cluster_)); });
  
    for (auto& t : threads) {
      t.join();
    }
  
    LOG(INFO) << "Insertion of records using threads has completed.";
  
    // Count the number of INSERTS.
    uint32_t inserts_snapshot = 0;
    while (true) {
      GetChangesResponsePB change_resp_after_snapshot =
          ASSERT_RESULT(UpdateCheckpoint(stream_id, tablets, &change_resp));
      uint32_t record_size_after_snapshot = change_resp_after_snapshot.cdc_sdk_proto_records_size();
      if (record_size_after_snapshot == 0) {
        break;
      }
      uint32_t insert_count = 0;
      for (uint32_t i = 0; i < record_size_after_snapshot; ++i) {
        const CDCSDKProtoRecordPB record = change_resp_after_snapshot.cdc_sdk_proto_records(i);
        if (record.row_message().op() == RowMessage::INSERT) {
          insert_count++;
        }
      }
      inserts_snapshot += insert_count;
      change_resp = change_resp_after_snapshot;
    }
    LOG(INFO) << "Got " << reads_snapshot + inserts_snapshot << " total (read + insert) record";
    ASSERT_EQ(reads_snapshot + inserts_snapshot, 20000);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(DropDatabase)) {
    ASSERT_OK(SetUpWithParams(3, 1, false));
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    ASSERT_OK(DropDB(&test_cluster_));
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestNeedSchemaInfoFlag)) {
    ASSERT_OK(SetUpWithParams(1, 1, false));
  
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
  
    google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
    ASSERT_OK(test_client()->GetTablets(table, 0, &tablets, /* partition_list_version = */ nullptr));
  
    std::string table_id = ASSERT_RESULT(GetTableId(&test_cluster_, kNamespaceName, kTableName));
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
  
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
    // This will write one row with PK = 0.
    ASSERT_OK(WriteRows(0 /* start */, 1 /* end */, &test_cluster_));
  
    // This is the first call to GetChanges, we will get a DDL record.
    auto resp = ASSERT_RESULT(VerifyIfDDLRecordPresent(stream_id, tablets, false, true));
  
    // Write another row to the database with PK = 1.
    ASSERT_OK(WriteRows(1 /* start */, 2 /* end */, &test_cluster_));
  
    // We will not get any DDL record here since this is not the first call and the flag
    // need_schema_info is also unset.
    resp = ASSERT_RESULT(
        VerifyIfDDLRecordPresent(stream_id, tablets, false, false, &resp.cdc_sdk_checkpoint()));
  
    // Write another row to the database with PK = 2.
    ASSERT_OK(WriteRows(2 /* start */, 3 /* end */, &test_cluster_));
  
    // We will get a DDL record since we have enabled the need_schema_info flag.
    resp = ASSERT_RESULT(
        VerifyIfDDLRecordPresent(stream_id, tablets, true, false, &resp.cdc_sdk_checkpoint()));
  }
  
  // Insert a single row, truncate table, insert another row.
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestTruncateTable)) {
    ASSERT_OK(SetUpWithParams(1, 1, false));
  
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName));
  
    google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
    ASSERT_OK(test_client()->GetTablets(table, 0, &tablets, /* partition_list_version = */ nullptr));
  
    TableId table_id = ASSERT_RESULT(GetTableId(&test_cluster_, kNamespaceName, kTableName));
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream());
    auto set_resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(set_resp.has_error());
    ASSERT_OK(WriteRows(0 /* start */, 1 /* end */, &test_cluster_));
    ASSERT_OK(TruncateTable(&test_cluster_, {table_id}));
    ASSERT_OK(WriteRows(1 /* start */, 2 /* end */, &test_cluster_));
  
    // Calling Get Changes without enabling truncate flag.
    // Expected records: (DDL, INSERT, INSERT).
    GetChangesResponsePB resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count_truncate_disable[] = {1, 2, 0, 0, 0, 0};
    uint32_t count_truncate_disable[] = {0, 0, 0, 0, 0, 0};
    ExpectedRecord expected_records_truncate_disable[] = {{0, 0}, {0, 1}, {1, 2}};
    uint32_t record_size = resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records_truncate_disable[i], count_truncate_disable);
    }
    CheckCount(expected_count_truncate_disable, count_truncate_disable);
  
    // Setting the flag true and calling Get Changes. This will enable streaming of truncate record.
    // Expected records: (DDL, INSERT, TRUNCATE, INSERT).
    FLAGS_stream_truncate_record = true;
    resp = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
  
    // The count array stores counts of DDL, INSERT, UPDATE, DELETE, READ, TRUNCATE in that order.
    const uint32_t expected_count_truncate_enable[] = {1, 2, 0, 0, 0, 1};
    uint32_t count_truncate_enable[] = {0, 0, 0, 0, 0, 0};
    ExpectedRecord expected_records_truncate_enable[] = {{0, 0}, {0, 1}, {0, 0}, {1, 2}};
    record_size = resp.cdc_sdk_proto_records_size();
    for (uint32_t i = 0; i < record_size; ++i) {
      const CDCSDKProtoRecordPB record = resp.cdc_sdk_proto_records(i);
      CheckRecord(record, expected_records_truncate_enable[i], count_truncate_enable);
    }
    CheckCount(expected_count_truncate_enable, count_truncate_enable);
  
    LOG(INFO) << "Got " << count_truncate_enable[0] << " ddl records, " << count_truncate_enable[1]
              << " insert records and " << count_truncate_enable[2] << " truncate records";
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestGarbageCollectionFlag)) {
    TestIntentGarbageCollectionFlag(1, true, 200);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestGarbageCollectionWithSmallInterval)) {
    TestIntentGarbageCollectionFlag(3, true, 200);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestGarbageCollectionWithLargerInterval)) {
    TestIntentGarbageCollectionFlag(3, true, 3000);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestNoGarbageCollectionBeforeInterval)) {
    TestIntentGarbageCollectionFlag(3, false, 0);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestSetCDCCheckpoint)) {
    TestSetCDCCheckpoint(1, false);
  }
  
  TEST_F(CDCSDKYsqlTest, YB_DISABLE_TEST_IN_TSAN(TestCheckPointPersistencyNodeRestart)) {
    FLAGS_enable_update_local_peer_min_index = false;
    FLAGS_update_min_cdc_indices_interval_secs = 1;
    FLAGS_cdc_state_checkpoint_update_interval_ms = 1;
    ASSERT_OK(SetUpWithParams(3, 1, false));
  
    const uint32_t num_tablets = 1;
    auto table = ASSERT_RESULT(CreateTable(&test_cluster_, kNamespaceName, kTableName, num_tablets));
    google::protobuf::RepeatedPtrField<master::TabletLocationsPB> tablets;
    ASSERT_OK(test_client()->GetTablets(table, 0, &tablets, /* partition_list_version =*/nullptr));
    ASSERT_EQ(tablets.size(), num_tablets);
  
    std::string table_id = ASSERT_RESULT(GetTableId(&test_cluster_, kNamespaceName, kTableName));
    CDCStreamId stream_id = ASSERT_RESULT(CreateDBStream(IMPLICIT));
  
    auto resp = ASSERT_RESULT(SetCDCCheckpoint(stream_id, tablets));
    ASSERT_FALSE(resp.has_error());
  
    // insert some records in transaction.
    ASSERT_OK(WriteRowsHelper(0 /* start */, 100 /* end */, &test_cluster_, true));
    ASSERT_OK(test_client()->FlushTables(
        {table.table_id()}, /* add_indexes = */ false, /* timeout_secs = */ 30,
        /* is_compaction = */ false));
  
    // call get changes.
    GetChangesResponsePB change_resp_1 = ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets));
    uint32_t record_size = change_resp_1.cdc_sdk_proto_records_size();
    LOG(INFO) << "Total records read by get change call: " << record_size;
  
    ASSERT_OK(WriteRowsHelper(100 /* start */, 200 /* end */, &test_cluster_, true));
    ASSERT_OK(test_client()->FlushTables(
        {table.table_id()}, /* add_indexes = */ false, /* timeout_secs = */ 30,
        /* is_compaction = */ false));
    // Greater than 100 check because  we got records for BEGIN, COMMIT also.
    ASSERT_GT(record_size, 100);
  
    // call get changes.
    GetChangesResponsePB change_resp_2 =
        ASSERT_RESULT(GetChangesFromCDC(stream_id, tablets, &change_resp_1.cdc_sdk_checkpoint()));
    record_size = change_resp_2.cdc_sdk_proto_records_size();
    ASSERT_GT(record_size, 100);
    LOG(INFO) << "Total records read by get change call: " << record_size;
  
    // Restart one of the node.
    SleepFor(MonoDelta::FromSeconds(1));
    test_cluster()->mini_tablet_server(1)->Shutdown();
    ASSERT_OK(test_cluster()->mini_tablet_server(1)->Start());
    ASSERT_OK(test_cluster()->mini_tablet_server(1)->WaitStarted());
  
    // Check all the tserver checkpoint info it's should be valid.
    for (size_t i = 0; i < test_cluster()->num_tablet_servers(); ++i) {
      for (const auto& peer : test_cluster()->GetTabletPeers(i)) {
        if (peer->tablet_id() == tablets[0].tablet_id()) {
          // What ever checkpoint persisted in the RAFT logs should be same as what ever in memory
          // transaction participant tablet peer.
          ASSERT_EQ(
              peer->cdc_sdk_min_checkpoint_op_id(),
              peer->tablet()->transaction_participant()->GetRetainOpId());
        }
      }
    }
  }
  
  }  // namespace enterprise
  }  // namespace cdc
  }  // namespace yb
  