[BACKPORT 2024.1][#22876][#22835][#22773] CDCSDK: Add new auto flag t…

…o identify non-eligible tables in CDC stream

Summary:
**Backport Description:**
No merge conflicts

**Original Description:**
Original commit: None / D36240
This diff is an extension of [[ https://phorge.dev.yugabyte.com/D36031 | D36031 ]] which introduced cleanup mechanism for non-eligible tables. The mechanism involves two steps:

  # Identification of indexes -> happens during loading of CDC streams into memory on a master restart/leadership change.
  # Removal of these identified indexes by the bg thread.

Without this diff, both these steps were guarded under a non-auto flag - `enable_cleanup_of_non_eligible_tables_from_cdcsdk_stream`
Therefore, post upgrade, step-1 requires the user to set the above flag and explicitly do a master restart/leader change.

To avoid this explicit master restart/leader change and still give control to users over this cleanup, we are introducing a new auto flag `cdcsdk_enable_identification_of_non_eligible_tables` that will guard the identification step.

These identified tables will be added to `namespace_to_cdcsdk_non_eligible_table_map_`.  If `enable_cleanup_of_non_eligible_tables_from_cdcsdk_stream` is set to true, the catalog manager background thread will pick up these tables for actual cleanup.
Jira: DB-11778, DB-11733, DB-11676

Test Plan:
Jenkins: urgent
Existing cdc tests for removal of non-eligible tables
./yb_build.sh --cxx-test integration-tests_cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestNonEligibleTableRemovalFromNonConsistentSnapshotCDCStream
./yb_build.sh --cxx-test integration-tests_cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestNonEligibleTableRemovalFromConsistentSnapshotCDCStream

./yb_build.sh --cxx-test integration-tests_cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestChildTabletsOfNonEligibleTableDoNotGetAddedToNonConsistentSnapshotStream
./yb_build.sh --cxx-test integration-tests_cdcsdk_ysql-test --gtest_filter CDCSDKYsqlTest.TestChildTabletsOfNonEligibleTableDoNotGetAddedToConsistentSnapshotStream

Reviewers: xCluster, hsunder, asrinivasan

Reviewed By: asrinivasan

Subscribers: ybase

Tags: #jenkins-ready

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

src/yb/master/xrepl_catalog_manager.cc

@@ -160,6 +160,17 @@ DEFINE_RUNTIME_bool(cdcsdk_enable_cleanup_of_non_eligible_tables_from_stream, fa
     "materialised view etc. in their stream metadata and these tables will be marked for removal "
     "by catalog manager background thread.");
 
+DEFINE_RUNTIME_AUTO_bool(cdcsdk_enable_identification_of_non_eligible_tables,
+                         kLocalPersisted,
+                         false,
+                         true,
+                         "This flag, when true, identifies all non-eligible tables that are part of"
+                         " a CDC stream metadata while loading the CDC streams on a master "
+                         "restart/leadership change. This identification happens on all CDC "
+                         "streams in the universe");
+TAG_FLAG(cdcsdk_enable_identification_of_non_eligible_tables, advanced);
+TAG_FLAG(cdcsdk_enable_identification_of_non_eligible_tables, hidden);
+
 DECLARE_bool(xcluster_wait_on_ddl_alter);
 DECLARE_int32(master_rpc_timeout_ms);
 DECLARE_bool(ysql_yb_enable_replication_commands);
@@ -342,7 +353,7 @@ class CDCStreamLoader : public Visitor<PersistentCDCStreamInfo> {
 
       // Check for any non-eligible tables like indexes, matview etc in CDC stream only if the
       // stream is not associated with a replication slot.
-      if (FLAGS_cdcsdk_enable_cleanup_of_non_eligible_tables_from_stream &&
+      if (FLAGS_cdcsdk_enable_identification_of_non_eligible_tables &&
           stream->GetCdcsdkYsqlReplicationSlotName().empty()) {
         catalog_manager_->FindAllNonEligibleTablesInCDCSDKStream(
             stream_id, metadata.table_id(), eligible_tables_info);
@@ -8007,7 +8018,7 @@ CatalogManager::UpdateCheckpointForTabletEntriesInCDCState(
         "from CDC stream $1",
         table_to_be_removed, stream_id);
 
-    LOG_WITH_FUNC(INFO) << "Deleting cdc state table entry (tablet,stream) - "
+    LOG_WITH_FUNC(INFO) << "Deleting cdc state table entry (tablet, stream, table) - "
                         << cdc_state_entries_to_be_deleted[0].ToString();
     RETURN_NOT_OK_PREPEND(
         cdc_state_table_->DeleteEntries(cdc_state_entries_to_be_deleted),