[#28033] CDC: Propagate the GetChanges() errors back to client

Summary:
Call to GetConsistentChanges() suppresses all the error status it gets from virtual_wal's GetConsistentChangesInternal() apart from few ones (such as 'stream being inactive'
and 'intent records already GCed'). This behaviour of not responding most of the errors back to walsender causes problems such as walsender being unaware of internal issue
keep trying and thus being stuck.
This diff corrects this issue by sending back most of the errors back to walsender which GetConsistentChanges() gets from virtual_wal's GetConsistentChangesInternal(). The error
cases which are skipped from propagating back to client are: when tablet peer is not started yet, and when the tablet is not in kAvailable state.

Jira: DB-17654, DB-16451

Test Plan:
./yb_build.sh --cxx-test cdcsdk_consumption_consistent_changes-test --gtest_filter=CDCSDKConsumptionConsistentChangesTest.TestBeforeImageNotExistErrorPropagation
./yb_build.sh release --cxx-test cdcsdk_consumption_consistent_changes-test --gtest_filter=CDCSDKConsumptionConsistentChangesTest.TestBeforeImageNotExistErrorPropagation

Reviewers: sumukh.phalgaonkar, skumar, stiwary, asrinivasan, #db-approvers

Reviewed By: sumukh.phalgaonkar, #db-approvers

Subscribers: ycdcxcluster

Tags: #jenkins-ready

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

Author: devansh5398

src/yb/cdc/cdc_service.cc

@@ -206,6 +206,10 @@ DEFINE_RUNTIME_bool(cdc_enable_implicit_checkpointing, false,
 DEFINE_RUNTIME_uint32(cdc_max_virtual_wal_per_tserver, 5,
                       "Maximum VirtualWAL instances that can be present on a tserver at any time.");
 
+DEFINE_test_flag(bool, mimic_tablet_not_in_available_state, false,
+    "If true, this is used to mimic the behavior of the tablet as if it is not in an available "
+    "state.");
+
 DECLARE_int32(log_min_seconds_to_retain);
 
 static bool ValidateMaxRefreshInterval(const char* flag_name, uint32 value) {
@@ -1601,6 +1605,13 @@ void CDCServiceImpl::GetChanges(
         status.IsTabletSplit() ? CDCErrorPB::TABLET_SPLIT : CDCErrorPB::INVALID_REQUEST, context);
   }
 
+  // Mocking that the tablet peer is not in TabletObjectState::kAvailable state.
+  if (PREDICT_FALSE(FLAGS_TEST_mimic_tablet_not_in_available_state)) {
+    RPC_STATUS_RETURN_ERROR(
+        STATUS(IllegalState, "Tablet not running: tablet object has invalid state"),
+        resp->mutable_error(), CDCErrorPB::INTERNAL_ERROR, context);
+  }
+
   auto tablet_peer = context_->LookupTablet(req->tablet_id());
 
   auto original_leader_term = tablet_peer ? tablet_peer->LeaderTerm() : OpId::kUnknownTerm;
@@ -4989,25 +5000,6 @@ Result<tablet::TabletPeerPtr> CDCServiceImpl::GetServingTablet(const TabletId& t
   return context_->GetServingTablet(tablet_id);
 }
 
-bool IsStreamInactiveError(Status status) {
-  if (!status.ok() && status.IsInternalError() &&
-      status.message().ToBuffer().find("expired for Tablet") != std::string::npos) {
-    return true;
-  }
-
-  return false;
-}
-
-bool IsIntentGCError(Status status) {
-  if (!status.ok() && status.IsInternalError() &&
-      status.message().ToBuffer().find("CDCSDK Trying to fetch already GCed intents") !=
-          std::string::npos) {
-    return true;
-  }
-
-  return false;
-}
-
 Status CDCServiceImpl::PersistActivePidInSlotEntry(
     const xrepl::StreamId& stream_id, uint64_t active_pid) {
   cdc::CDCStateTableEntry entry(kCDCSDKSlotEntryTabletId, stream_id);
@@ -5239,11 +5231,7 @@ void CDCServiceImpl::GetConsistentChanges(
         Format("GetConsistentChanges failed for stream_id: $0 with error: $1", stream_id, s);
     if (!s.IsTryAgain()) {
       LOG(WARNING) << msg;
-      // Propogate the error to the client only when the stream has expired or the intents have been
-      // garbage collected.
-      if (IsStreamInactiveError(s) || IsIntentGCError(s)) {
-        RPC_STATUS_RETURN_ERROR(s, resp->mutable_error(), CDCErrorPB::INTERNAL_ERROR, context);
-      }
+      RPC_STATUS_RETURN_ERROR(s, resp->mutable_error(), CDCErrorPB::INTERNAL_ERROR, context);
     } else {
       YB_LOG_EVERY_N_SECS(WARNING, 300) << msg;
     }

src/yb/cdc/cdcsdk_producer.cc

@@ -588,16 +588,17 @@ Status PopulateBeforeImage(
       tablet_peer, commit_time.Decremented(), row_message, cdc_sdk_safe_time,
       std::forward<Args>(args)...);
   if (!status.ok()) {
-    LOG(DFATAL) << "Failed to get the BeforeImage for tablet: " << tablet_peer->tablet_id()
-                << " with read time: " << commit_time
-                << " for change record type: " << row_message->op()
-                << " row_message: " << row_message->DebugString()
-                << " with error status: " << status;
+    LOG(WARNING) << "Failed to get the BeforeImage for tablet: " << tablet_peer->tablet_id()
+                 << " with read time: " << commit_time
+                 << " for change record type: " << row_message->op()
+                 << " row_message: " << row_message->DebugString()
+                 << " with error status: " << status;
+  } else {
+    VLOG(2) << "Successfully got the BeforeImage for tablet: " << tablet_peer->tablet_id()
+            << " with read time: " << commit_time
+            << " for change record type: " << row_message->op()
+            << " row_message: " << row_message->DebugString();
   }
-  VLOG(2) << "Successfully got the BeforeImage for tablet: " << tablet_peer->tablet_id()
-          << " with read time: " << commit_time
-          << " for change record type: " << row_message->op()
-          << " row_message: " << row_message->DebugString();
   return status;
 }
 

src/yb/cdc/cdcsdk_virtual_wal.cc

@@ -840,6 +840,30 @@ Status CDCSDKVirtualWAL::GetConsistentChangesInternal(
   return Status::OK();
 }
 
+bool IsRetryableError(const Status& status) {
+  // The following error cases are considered retryable:
+  // - when the tablet peer is not started yet, or
+  // - the tablet is not in available state.
+  // If any additional errors need to be made retryable in VWAL, add them to the list above.
+
+  DCHECK(!status.ok()) << "Status is not expected to be OK when calling IsRetryableError, "
+                       << "status: " << status.ToString();
+
+  // Tablet peer is not started yet
+  if (status.IsIllegalState() &&
+      status.message().ToBuffer().find("is not started yet") != std::string::npos) {
+    return true;
+  }
+
+  // Tablet is not in kAvailable state
+  if (status.IsIllegalState() &&
+      status.message().ToBuffer().find("Tablet not running") != std::string::npos) {
+    return true;
+  }
+
+  return false;
+}
+
 Status CDCSDKVirtualWAL::GetChangesInternal(
     const std::unordered_set<TabletId> tablet_to_poll_list, HostPort hostport,
     CoarseTimePoint deadline) {
@@ -885,6 +909,11 @@ Status CDCSDKVirtualWAL::GetChangesInternal(
           }
           continue;
         } else {
+          // Replace the status code with 'kTryAgain' for the errors which are retryable so they
+          // don't get propagated to walsender.
+          if (IsRetryableError(s)) {
+            s = s.CloneAndReplaceCode(Status::Code::kTryAgain);
+          }
           LOG_WITH_PREFIX(WARNING) << "GetChanges failed for tablet_id: " << tablet_id
                                    << " with error: " << s.CloneAndPrepend(error_msg).ToString();
           RETURN_NOT_OK(s);

src/yb/integration-tests/cdcsdk_consumption_consistent_changes-test.cc

@@ -2682,6 +2682,49 @@ TEST_F(
   ASSERT_EQ(tablet_peer_2->get_cdc_sdk_safe_time(), slot_row->record_id_commit_time);
 }
 
+TEST_F(CDCSDKConsumptionConsistentChangesTest, TestBeforeImageNotExistErrorPropagation) {
+  ASSERT_OK(SetUpWithParams(1, 1, false));
+  uint32_t num_cols = 3;
+  auto table = ASSERT_RESULT(CreateTable(
+    &test_cluster_, kNamespaceName, kTableName, 1, true, false, 0, false, "", "public", num_cols));
+
+  auto conn = ASSERT_RESULT(test_cluster_.ConnectToDB(kNamespaceName));
+  // Set the replica identity to FULL. This is needed to get before image in update operation.
+  ASSERT_OK(conn.ExecuteFormat("ALTER TABLE $0 REPLICA IDENTITY FULL", kTableName));
+
+  auto stream_id = ASSERT_RESULT(CreateConsistentSnapshotStreamWithReplicationSlot());
+  ASSERT_OK(InitVirtualWAL(stream_id, {table.table_id()}));
+
+  // Setting the flag to mimic tablet not in available state. The expectation is that
+  // CDCServiceImpl::GetConsistentChanges() should not return an error since such error is expected
+  // to be retryable.
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_TEST_mimic_tablet_not_in_available_state) = true;
+  ASSERT_OK(GetConsistentChangesFromCDC(stream_id));
+
+  // Resetting the test flag.
+  ANNOTATE_UNPROTECTED_WRITE(FLAGS_TEST_mimic_tablet_not_in_available_state) = false;
+
+  map<std::string, uint32_t> col_val_map1, col_val_map2;
+  col_val_map1.insert({"col2", 9});
+  col_val_map1.insert({"col3", 10});
+  col_val_map2.insert({"col2", 10});
+  col_val_map2.insert({"col3", 11});
+  ASSERT_OK(UpdateRowsHelper(1, 2, &test_cluster_, true, 1, col_val_map1, col_val_map2, num_cols));
+
+  // Getting the consistent changes from stream and expecting a non-ok status due to "Failed to get
+  // the beforeimage" error. This error occurs when an UPDATE follows an INSERT for the same row in
+  // the same transaction, and the before image is unavailable.
+  // However, when packed rows are enabled, all the UPDATE operation/s of the same row gets absorbed
+  // into the INSERT operation and hence the before image is not required. In that case, we expect
+  // GetConsistentChanges to return the records without any error.
+  if (FLAGS_ysql_enable_packed_row) {
+    ASSERT_OK(GetConsistentChangesFromCDC(stream_id));
+  } else {
+    ASSERT_NOK_STR_CONTAINS(
+        GetConsistentChangesFromCDC(stream_id), "Failed to get the beforeimage");
+  }
+}
+
 // This test verifies the behaviour of VWAL when the publication refresh interval is changed when
 // consumption is in progress.
 TEST_F(CDCSDKConsumptionConsistentChangesTest, TestChangingPublicationRefreshInterval) {

src/yb/integration-tests/cdcsdk_ysql_test_base.cc

@@ -1560,14 +1560,7 @@ Result<GetConsistentChangesResponsePB> CDCSDKYsqlTest::GetConsistentChangesFromC
 
         if (status.ok() && change_resp.has_error()) {
           status = StatusFromPB(change_resp.error().status());
-          if (status.IsNotFound() || status.IsInvalidArgument()) {
-            RETURN_NOT_OK(status);
-          } else if (status.IsInternalError()) {
-            auto err_msg = status.message().ToBuffer();
-            if ((err_msg.find("expired for Tablet") ||
-                  err_msg.find("CDCSDK Trying to fetch already GCed intents")))
-              RETURN_NOT_OK(status);
-          }
+          RETURN_NOT_OK(status);
         }
 
         return false;

src/yb/integration-tests/cdcsdk_ysql_test_base.h

@@ -139,6 +139,7 @@ DECLARE_int32(retryable_request_timeout_secs);
 DECLARE_bool(save_index_into_wal_segments);
 DECLARE_bool(TEST_skip_process_apply);
 DECLARE_bool(TEST_ysql_yb_enable_implicit_dynamic_tables_logical_replication);
+DECLARE_bool(TEST_mimic_tablet_not_in_available_state);
 
 namespace yb {