[BACKPORT 2024.1][yugabyte#19294] CDCSDK: Refactor reading of WAL messages for consistent CDC

Summary:
####Backport Description
Minor merge conflict was encountered in cdcsdk_test_base.h at the declaration of test flag.

####Original Description
In a workload with single shard transactions, it was observed that WAL contained operations that were replicated but not committed. In such a scenario an empty batch is returned by `ReadReplicatedMessagesForCDC` if the from_op_id is same as the commited_op_id. CDCSDK producer ends up updating the safe time. Hence in the next GetChanges call, the previously uncommitted messages get filtered out. This can lead to data loss in CDC operations.

This diff introduces a new method `ReadReplicatedMessagesForConsistentCDC` which takes consistent_stream_safe_time as a parameter and reads the WAL until the committed_op_id or till we encounter a message with hybrid time greater than consistent_stream_safe_time. This method checks whether committed_op_id and majority_replicated_op_id are same. If they differ, it waits in a loop until both become equal. However in the case where we reach close to the deadline and the mismatch between committed_op_id and majority_replicated_op_id still exists, in order to prevent GetChanges rpc timeout `ReadReplicatedMessagesForConsistentCDC` will return an empty batch of messages with an indication about the mismatch. In such a scenario cdcsdk producer will not update its safe time. The GetChanges response will have the same safe_hybrid_time as the request in this case.
Jira: DB-8105

Original commit: a0264d7 / D30420

Test Plan:
Existing CDC tests and QA tests
./yb_build.sh --cxx-test integration-tests_cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestWithMajorityReplicatedButNonCommittedSingleShardTxn

Reviewers: vkushwaha, skumar, sergei, asrinivasan, stiwary, siddharth.shah

Reviewed By: siddharth.shah

Subscribers: ycdcxcluster, ybase, bogdan, sergei

Tags: #jenkins-ready

Differential Revision: https://phorge.dev.yugabyte.com/D37733

Author: Sumukh-Phalgaonkar

src/yb/cdc/cdcsdk_producer.cc

@@ -1984,64 +1984,65 @@ Status GetConsistentWALRecords(
           << ", last_seen_op_id: " << last_seen_op_id->ToString()
           << ", historical_max_op_id: " << historical_max_op_id;
   auto consensus = VERIFY_RESULT(tablet_peer->GetConsensus());
-  do {
-    auto read_ops = VERIFY_RESULT(consensus->ReadReplicatedMessagesForCDC(
-        *last_seen_op_id, *last_readable_opid_index, deadline));
-
-    if (read_ops.messages.empty()) {
-      VLOG_WITH_FUNC(1) << "Did not get any messages with current batch of 'read_ops'."
-                        << "last_seen_op_id: " << *last_seen_op_id << ", last_readable_opid_index "
-                        << **last_readable_opid_index;
-      break;
-    }
-
-    if (read_ops.read_from_disk_size && mem_tracker) {
-      (*consumption) = ScopedTrackedConsumption(mem_tracker, read_ops.read_from_disk_size);
-    }
+  // Read the committed WAL messages with hybrid time <= consistent_stream_safe_time. If there exist
+  // messages in the WAL which are replicated but not yet committed,
+  // ReadReplicatedMessagesForConsistentCDC waits for them to get committed and eventually includes
+  // them in the result.
+  auto read_ops = VERIFY_RESULT(consensus->ReadReplicatedMessagesForConsistentCDC(
+      *last_seen_op_id, consistent_safe_time, deadline, false, *last_readable_opid_index));
 
-    for (const auto& msg : read_ops.messages) {
-      last_seen_op_id->term = msg->id().term();
-      last_seen_op_id->index = msg->id().index();
-
-      if (IsIntent(msg) || (IsUpdateTransactionOp(msg) &&
-                            msg->transaction_state().status() != TransactionStatus::APPLYING)) {
-        continue;
-      }
+  if (read_ops.read_from_disk_size && mem_tracker) {
+    (*consumption) = ScopedTrackedConsumption(mem_tracker, read_ops.read_from_disk_size);
+  }
 
-      if (VLOG_IS_ON(3) && IsUpdateTransactionOp(msg) &&
-          msg->transaction_state().status() == TransactionStatus::APPLYING) {
-        auto txn_id =
-            VERIFY_RESULT(FullyDecodeTransactionId(msg->transaction_state().transaction_id()));
-        VLOG(3) << "Read transaction in WAL on "
-                << "tablet_id: " << tablet_peer->tablet_id() << ", transaction_id: " << txn_id
-                << ", OpId: " << msg->id().term() << "." << msg->id().index()
-                << ", commit_time: " << GetTransactionCommitTime(msg)
-                << ", consistent safe_time: " << consistent_safe_time
-                << ", safe_hybrid_time_req: " << safe_hybrid_time_req;
-      } else if (VLOG_IS_ON(3)) {
-        VLOG(3) << "Read WAL msg on "
-                << "tablet_id: " << tablet_peer->tablet_id() << ", op_type: " << msg->op_type()
-                << ", OpId: " << msg->id().term() << "." << msg->id().index()
-                << ", commit_time: " << GetTransactionCommitTime(msg)
-                << ", consistent safe_time: " << consistent_safe_time
-                << ", safe_hybrid_time_req: " << safe_hybrid_time_req;
-      }
+  for (const auto& msg : read_ops.messages) {
+    last_seen_op_id->term = msg->id().term();
+    last_seen_op_id->index = msg->id().index();
 
-      all_checkpoints->push_back(msg);
-      consistent_wal_records->push_back(msg);
+    if (IsIntent(msg) || (IsUpdateTransactionOp(msg) &&
+                          msg->transaction_state().status() != TransactionStatus::APPLYING)) {
+      continue;
     }
 
-    if (read_ops.messages.size() > 0) {
-      *msgs_holder = consensus::ReplicateMsgsHolder(
-          nullptr, std::move(read_ops.messages), std::move((*consumption)));
+    if (VLOG_IS_ON(3) && IsUpdateTransactionOp(msg) &&
+        msg->transaction_state().status() == TransactionStatus::APPLYING) {
+      auto txn_id =
+          VERIFY_RESULT(FullyDecodeTransactionId(msg->transaction_state().transaction_id()));
+      VLOG(3) << "Read transaction in WAL on "
+              << "tablet_id: " << tablet_peer->tablet_id() << ", transaction_id: " << txn_id
+              << ", OpId: " << msg->id().term() << "." << msg->id().index()
+              << ", commit_time: " << GetTransactionCommitTime(msg)
+              << ", consistent safe_time: " << consistent_safe_time
+              << ", safe_hybrid_time_req: " << safe_hybrid_time_req;
+    } else if (VLOG_IS_ON(3)) {
+      VLOG(3) << "Read WAL msg on "
+              << "tablet_id: " << tablet_peer->tablet_id() << ", op_type: " << msg->op_type()
+              << ", OpId: " << msg->id().term() << "." << msg->id().index()
+              << ", commit_time: " << GetTransactionCommitTime(msg)
+              << ", consistent safe_time: " << consistent_safe_time
+              << ", safe_hybrid_time_req: " << safe_hybrid_time_req;
     }
-  } while (((*last_readable_opid_index) && last_seen_op_id->index < **last_readable_opid_index));
+
+    all_checkpoints->push_back(msg);
+    consistent_wal_records->push_back(msg);
+  }
+
+  if (read_ops.messages.size() > 0) {
+    *msgs_holder = consensus::ReplicateMsgsHolder(
+        nullptr, std::move(read_ops.messages), std::move((*consumption)));
+  }
 
   // Handle the case where WAL doesn't have the apply record for all the committed transactions.
   if (historical_max_op_id.valid() && historical_max_op_id > *last_seen_op_id) {
     (*wait_for_wal_update) = true;
   }
 
+  if (consistent_wal_records->empty() && read_ops.have_more_messages) {
+    VLOG(1) << "Received empty read_ops with have_more_messages set to true, indicating presence "
+               "of replicated but not committed records in the WAL";
+    *wait_for_wal_update = true;
+  }
+
   SortConsistentWALRecords(consistent_wal_records);
   VLOG_WITH_FUNC(1) << "Got a total of " << consistent_wal_records->size() << " WAL records "
                     << "in the current segment";

src/yb/consensus/consensus.h

@@ -341,6 +341,11 @@ class Consensus {
       const yb::OpId& from, int64_t* repl_index, const CoarseTimePoint deadline,
       const bool fetch_single_entry = false) = 0;
 
+  // Read all the committed messages for CDC producer.
+  virtual Result<ReadOpsResult> ReadReplicatedMessagesForConsistentCDC(
+      OpId from, uint64_t stream_safe_time, CoarseTimePoint deadline,
+      bool fetch_single_entry = false, int64_t* repl_index = nullptr) = 0;
+
   virtual void UpdateCDCConsumerOpId(const yb::OpId& op_id) = 0;
 
  protected:

src/yb/consensus/consensus_queue.cc

@@ -134,6 +134,13 @@ DEFINE_RUNTIME_uint32(max_remote_bootstrap_attempts_from_non_leader, 5,
 DEFINE_test_flag(bool, assert_remote_bootstrap_happens_from_same_zone, false,
     "Assert that remote bootstrap is served by a peer in the same zone as the new peer.");
 
+DEFINE_test_flag(bool, stop_committed_op_id_updation, false,
+    "Test flag to stop the updation of committed_op_id");
+
+DEFINE_RUNTIME_uint32(cdcsdk_wal_reads_deadline_buffer_secs, 5,
+    "This flag determines the buffer time from the deadline at which we must stop reading the WAL "
+    "messages and start processing the records we have read till now.");
+
 namespace yb {
 namespace consensus {
 
@@ -729,6 +736,25 @@ Result<ReadOpsResult> PeerMessageQueue::ReadFromLogCache(
   return result;
 }
 
+Result<ReadOpsResult> PeerMessageQueue::ReadFromLogCacheForCDC(
+    OpId last_op_id, int64_t to_index, CoarseTimePoint deadline, bool fetch_single_entry) {
+  // If an empty OpID is only sent on the first read request, start at the earliest known entry.
+  int64_t after_op_index =
+      last_op_id.empty() ? max<int64_t>(log_cache_.earliest_op_index(), 0)
+                         : last_op_id.index;
+
+  auto result = ReadFromLogCache(
+      after_op_index, to_index, FLAGS_consensus_max_batch_size_bytes, local_peer_uuid_, deadline,
+      fetch_single_entry);
+  if (PREDICT_FALSE(!result.ok()) && PREDICT_TRUE(result.status().IsNotFound())) {
+    const std::string premature_gc_warning = Format(
+        "The logs from index $0 have been garbage collected and cannot be read ", after_op_index);
+    LOG_WITH_PREFIX(WARNING) << premature_gc_warning;
+    return result.status().CloneAndPrepend(premature_gc_warning);
+  }
+  return result;
+}
+
 // Read majority replicated messages from cache for CDC.
 // CDC producer will use this to get the messages to send in response to cdc::GetChanges RPC.
 Result<ReadOpsResult> PeerMessageQueue::ReadReplicatedMessagesForCDC(
@@ -751,31 +777,95 @@ Result<ReadOpsResult> PeerMessageQueue::ReadReplicatedMessagesForCDC(
 
   if (last_op_id.index >= to_index && !fetch_single_entry) {
     // Nothing to read.
-    return ReadOpsResult();
+    return ReadOpsResult {
+      .messages = ReplicateMsgs(),
+      .preceding_op = OpId(),
+      .have_more_messages = HaveMoreMessages(pending_messages)
+    };
   }
 
-  // If an empty OpID is only sent on the first read request, start at the earliest known entry.
-  int64_t after_op_index = last_op_id.empty() ?
-                             max(log_cache_.earliest_op_index(), last_op_id.index) :
-                             last_op_id.index;
+  auto result =
+      VERIFY_RESULT(ReadFromLogCacheForCDC(last_op_id, to_index, deadline, fetch_single_entry));
+
+  result.have_more_messages =
+      HaveMoreMessages(result.have_more_messages.get() || pending_messages);
 
-  auto result = ReadFromLogCache(
-      after_op_index, to_index, FLAGS_consensus_max_batch_size_bytes, local_peer_uuid_, deadline,
-      fetch_single_entry);
-  if (PREDICT_FALSE(!result.ok()) && PREDICT_TRUE(result.status().IsNotFound())) {
-    const std::string premature_gc_warning =
-      Format("The logs from index $0 have been garbage collected and cannot be read ($1)",
-             after_op_index, result.status());
-    LOG_WITH_PREFIX(INFO) << premature_gc_warning;
-    return STATUS(NotFound, premature_gc_warning);
-  }
-  if (result.ok()) {
-    result->have_more_messages = HaveMoreMessages(result->have_more_messages.get() ||
-                                                  pending_messages);
-  }
   return result;
 }
 
+// Read all the commited messages from cache for CDC.
+// CDC producer will use these to get the messages to send in response to cdc::GetChanges RPC.
+Result<ReadOpsResult> PeerMessageQueue::ReadReplicatedMessagesForConsistentCDC(
+    OpId last_op_id, uint64_t stream_safe_time, CoarseTimePoint deadline, bool fetch_single_entry,
+    int64_t* repl_index) {
+  auto res = ReadOpsResult();
+  res.have_more_messages = HaveMoreMessages(false);
+  int64_t committed_op_id_index;
+  int64_t last_replicated_op_id_index;
+  bool pending_messages = false;
+  uint64_t last_read_hybrid_time = 0;
+
+  do {
+    // Return if we reach close to the deadline, providing time for cdc producer and virtual WAL
+    // to process the records.
+    if (deadline - CoarseMonoClock::Now() <= FLAGS_cdcsdk_wal_reads_deadline_buffer_secs * 1s) {
+      return res;
+    }
+
+    {
+      LockGuard lock(queue_lock_);
+      // Use committed_op_id because it's already been processed by the Transaction codepath.
+      committed_op_id_index = queue_state_.committed_op_id.index;
+      // Determine if there are pending operations in RAFT but not yet LogCache.
+      last_replicated_op_id_index = queue_state_.majority_replicated_op_id.index;
+      pending_messages = committed_op_id_index != last_replicated_op_id_index;
+    }
+
+    if (repl_index) {
+      *repl_index = committed_op_id_index;
+    }
+
+    if (last_op_id.index >= committed_op_id_index && !fetch_single_entry) {
+      if (pending_messages) {
+        // Wait for committed_op_id to match majority_replicated_op_id.
+        res.have_more_messages = HaveMoreMessages(pending_messages);
+        continue;
+      } else {
+        // Nothing to read.
+        return ReadOpsResult{
+            .messages = ReplicateMsgs(),
+            .preceding_op = last_op_id,
+            .have_more_messages = HaveMoreMessages::kFalse};
+      }
+    }
+
+    auto result = VERIFY_RESULT(
+        ReadFromLogCacheForCDC(last_op_id, committed_op_id_index, deadline, fetch_single_entry));
+
+    res.messages.insert(res.messages.end(), result.messages.begin(), result.messages.end());
+    res.read_from_disk_size += result.read_from_disk_size;
+    pending_messages |= result.have_more_messages.get();
+    res.have_more_messages = HaveMoreMessages(pending_messages);
+
+    if (res.messages.size() > 0) {
+      auto msg = res.messages.back();
+      last_op_id = OpId::FromPB(msg->id());
+      last_read_hybrid_time = msg->hybrid_time();
+    } else {
+      // If an empty last_op_id is sent in the first read request, then ReadFromLogCacheForCDC reads
+      // from the earliest known OpId. If this earliest known OpId turns out to be same as
+      // committed_op_id then we receive an empty message list in the result. The earliest known
+      // OpId is present in the preceding_op of the result. We update the last_op_id with this to
+      // prevent unncessary looping.
+      last_op_id = result.preceding_op;
+    }
+
+  } while ((last_op_id.index < committed_op_id_index || pending_messages) &&
+           last_read_hybrid_time <= stream_safe_time);
+
+  return res;
+}
+
 const PeerMessageQueue::TrackedPeer* PeerMessageQueue::FindClosestPeerForBootstrap(
     const TrackedPeer* remote_tracked_peer) {
   const CloudInfoPB& src_cloud_info = remote_tracked_peer->cloud_info.value();
@@ -1602,7 +1692,8 @@ void PeerMessageQueue::NotifyObserversOfMajorityReplOpChangeTask(
   {
     LockGuard lock(queue_lock_);
     if (!new_committed_op_id.empty() &&
-        new_committed_op_id.index > queue_state_.committed_op_id.index) {
+        new_committed_op_id.index > queue_state_.committed_op_id.index &&
+        !GetAtomicFlag(&FLAGS_TEST_stop_committed_op_id_updation)) {
       queue_state_.committed_op_id = new_committed_op_id;
     }
     queue_state_.last_applied_op_id.MakeAtLeast(last_applied_op_id);

src/yb/consensus/consensus_queue.h

@@ -401,6 +401,10 @@ class PeerMessageQueue {
       const CoarseTimePoint deadline = CoarseTimePoint::max(),
       const bool fetch_single_entry = false);
 
+  Result<ReadOpsResult> ReadReplicatedMessagesForConsistentCDC(
+      OpId last_op_id, uint64_t stream_safe_time, CoarseTimePoint deadline,
+      bool fetch_single_entry = false, int64_t* last_replicated_opid_index = nullptr);
+
   void UpdateCDCConsumerOpId(const yb::OpId& op_id);
 
   // Get the maximum op ID that can be evicted for CDC consumer from log cache.
@@ -578,6 +582,12 @@ class PeerMessageQueue {
       const CoarseTimePoint deadline = CoarseTimePoint::max(),
       const bool fetch_single_entry = false);
 
+  Result<ReadOpsResult> ReadFromLogCacheForCDC(
+      OpId last_op_id,
+      int64_t to_index,
+      CoarseTimePoint deadline = CoarseTimePoint::max(),
+      bool fetch_single_entry = false);
+
   void TEST_WaitForNotificationToFinish();
 
   std::vector<PeerMessageQueueObserver*> observers_;

src/yb/consensus/raft_consensus.cc

@@ -3640,6 +3640,13 @@ Result<ReadOpsResult> RaftConsensus::ReadReplicatedMessagesForCDC(
       from, last_replicated_opid_index, deadline, fetch_single_entry);
 }
 
+Result<ReadOpsResult> RaftConsensus::ReadReplicatedMessagesForConsistentCDC(
+    OpId from, uint64_t stream_safe_time, CoarseTimePoint deadline, bool fetch_single_entry,
+    int64_t* last_replicated_opid_index) {
+  return queue_->ReadReplicatedMessagesForConsistentCDC(
+      from, stream_safe_time, deadline, fetch_single_entry, last_replicated_opid_index);
+}
+
 void RaftConsensus::UpdateCDCConsumerOpId(const yb::OpId& op_id) {
   return queue_->UpdateCDCConsumerOpId(op_id);
 }

src/yb/consensus/raft_consensus.h

@@ -276,6 +276,13 @@ class RaftConsensus : public std::enable_shared_from_this<RaftConsensus>,
       const CoarseTimePoint deadline = CoarseTimePoint::max(),
       const bool fetch_single_entry = false) override;
 
+  Result<ReadOpsResult> ReadReplicatedMessagesForConsistentCDC(
+      OpId from,
+      uint64_t stream_safe_time,
+      CoarseTimePoint deadline,
+      bool fetch_single_entry = false,
+      int64_t* last_replicated_opid_index = nullptr) override;
+
   void UpdateCDCConsumerOpId(const yb::OpId& op_id) override;
 
   // Start memory tracking of following operation in case it is still present in our caches.

src/yb/integration-tests/cdcsdk_test_base.h

@@ -52,6 +52,7 @@ DECLARE_bool(ysql_enable_pack_full_row_update);
 DECLARE_bool(ysql_yb_enable_replica_identity);
 DECLARE_bool(ysql_enable_packed_row_for_colocated_table);
 DECLARE_bool(yb_enable_cdc_consistent_snapshot_streams);
+DECLARE_bool(TEST_stop_committed_op_id_updation);
 
 namespace yb {
 using client::YBClient;