[CDCSDK] [#9019] CDC SDK API changes

Summary:
Added following CDC admin APIs. The CDC stream is created at the namespace level. It creates the stream on all the tables present at that point in time.
This Stream ID can be used by the CDC client to get the data of the tables. Currently, any table created in this namespace after this command is executed, will not have the stream created for it automatically. The user will need to alter this stream to add/remove the table id.

**Case 1: No tables are present in the database**
```
$ ./yb-admin create_change_data_stream demo
CDC Stream ID: 3ea999960f5f4fe49d5526079cd0eec4

$ ./yb-admin list_change_data_streams
CDC Streams:
streams {
  stream_id: "3ea999960f5f4fe49d5526079cd0eec4"
  options {
    key: "id_type"
    value: "NAMESPACEID"
  }
  options {
    key: "checkpoint_type"
    value: "EXPLICIT"
  }
  options {
    key: "source_type"
    value: "CDCSDK"
  }
  options {
    key: "record_format"
    value: "PROTO"
  }
  options {
    key: "record_type"
    value: "CHANGE"
  }
  options {
    key: "state"
    value: "ACTIVE"
  }
}

$ ./yb-admin get_change_data_stream_info 3ea999960f5f4fe49d5526079cd0eec4
CDC DB Stream Info:
namespace_id: "000033e1000030008000000000000000"

$ ./yb-admin delete_change_data_stream 3ea999960f5f4fe49d5526079cd0eec4
Successfully deleted Change Data Stream ID: 3ea999960f5f4fe49d5526079cd0eec4
```

**Case 2: Some tables are present in the DB**
In this case, we will be getting the table IDs of the respective tables in the relevant responses
```
$ ./yb-admin create_change_data_stream demo
CDC Stream ID: a6e987dbc2af4516b02ab53dfd01cf56

$ ./yb-admin list_change_data_streams
CDC Streams:
streams {
  stream_id: "a6e987dbc2af4516b02ab53dfd01cf56"
  table_id: "000033e1000030008000000000004000"
  table_id: "000033e1000030008000000000004005"
  options {
    key: "id_type"
    value: "NAMESPACEID"
  }
  options {
    key: "checkpoint_type"
    value: "EXPLICIT"
  }
  options {
    key: "source_type"
    value: "CDCSDK"
  }
  options {
    key: "record_format"
    value: "PROTO"
  }
  options {
    key: "record_type"
    value: "CHANGE"
  }
  options {
    key: "state"
    value: "ACTIVE"
  }
}

$ ./yb-admin get_change_data_stream_info a6e987dbc2af4516b02ab53dfd01cf56
CDC DB Stream Info:
table_info {
  stream_id: "a6e987dbc2af4516b02ab53dfd01cf56"
  table_id: "000033e1000030008000000000004000"
}
table_info {
  stream_id: "a6e987dbc2af4516b02ab53dfd01cf56"
  table_id: "000033e1000030008000000000004005"
}
namespace_id: "000033e1000030008000000000000000"

$ ./yb-admin delete_change_data_stream a6e987dbc2af4516b02ab53dfd01cf56
Successfully deleted Change Data Stream ID: a6e987dbc2af4516b02ab53dfd01cf56
```

Test Plan: Added unit tests. Have tested it manually.

Reviewers: rahuldesirazu, nicolas, sergei, vkushwaha

Reviewed By: sergei, vkushwaha

Subscribers: zyu, ybase, bogdan

Differential Revision: https://phabricator.dev.yugabyte.com/D14605

Author: suranjan

ent/src/yb/cdc/CMakeLists-include.txt

@@ -25,7 +25,8 @@ set(CDC_SRCS_EXTENSIONS
   ${YB_ENT_CURRENT_SOURCE_DIR}/cdc_service.cc
   ${YB_ENT_CURRENT_SOURCE_DIR}/cdc_metrics.cc
   ${YB_ENT_CURRENT_SOURCE_DIR}/cdc_producer.cc
-  ${YB_ENT_CURRENT_SOURCE_DIR}/cdc_rpc.cc)
+  ${YB_ENT_CURRENT_SOURCE_DIR}/cdc_rpc.cc
+  ${YB_ENT_CURRENT_SOURCE_DIR}/cdc_error.cc)
 
 ADD_YB_LIBRARY(
     cdc

ent/src/yb/cdc/cdc_error.cc

@@ -0,0 +1,43 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+//
+// The following only applies to changes made to this file as part of YugaByte development.
+//
+// Portions 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 "yb/cdc/cdc_error.h"
+
+namespace yb {
+namespace cdc {
+
+static const std::string kCDCErrorCategoryName = "CDC error";
+
+static StatusCategoryRegisterer cdc_error_category_registerer(
+    StatusCategoryDescription::Make<CDCErrorTag>(&kCDCErrorCategoryName));
+
+} // namespace cdc
+} // namespace yb

ent/src/yb/cdc/cdc_error.h

@@ -0,0 +1,59 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you 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.
+//
+// The following only applies to changes made to this file as part of YugaByte development.
+//
+// Portions 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.
+//
+
+#ifndef ENT_SRC_YB_CDC_CDC_ERROR_H
+#define ENT_SRC_YB_CDC_CDC_ERROR_H
+
+#include "yb/cdc/cdc_service.pb.h"
+
+#include "yb/util/status_ec.h"
+
+namespace yb {
+namespace cdc {
+
+struct CDCErrorTag : IntegralErrorTag<CDCErrorPB::Code> {
+  // This category id is part of the wire protocol and should not be changed once released.
+  static constexpr uint8_t kCategory = 16;
+
+  typedef CDCErrorPB::Code Code;
+
+  static const std::string& ToMessage(Code code) {
+    return CDCErrorPB::Code_Name(code);
+  }
+};
+
+typedef StatusErrorCodeImpl<CDCErrorTag> CDCError;
+
+} // namespace cdc
+} // namespace yb
+
+#endif // ENT_SRC_YB_CDC_CDC_ERROR_H

ent/src/yb/cdc/cdc_producer.h

@@ -20,6 +20,7 @@
 #include <boost/unordered_map.hpp>
 
 #include "yb/cdc/cdc_service.service.h"
+#include "yb/common/common_fwd.h"
 #include "yb/common/transaction.h"
 #include "yb/consensus/consensus_fwd.h"
 #include "yb/docdb/docdb.pb.h"
@@ -34,14 +35,27 @@ class MemTracker;
 namespace cdc {
 
 struct StreamMetadata {
-  TableId table_id;
+  NamespaceId ns_id;
+  std::vector<TableId> table_ids;
   CDCRecordType record_type;
   CDCRecordFormat record_format;
+  CDCRequestSource source_type;
+  CDCCheckpointType checkpoint_type;
 
   StreamMetadata() = default;
 
-  StreamMetadata(TableId table_id, CDCRecordType record_type, CDCRecordFormat record_format)
-      : table_id(std::move(table_id)), record_type(record_type), record_format(record_format) {
+  StreamMetadata(NamespaceId ns_id,
+                 std::vector<TableId> table_ids,
+                 CDCRecordType record_type,
+                 CDCRecordFormat record_format,
+                 CDCRequestSource source_type,
+                 CDCCheckpointType checkpoint_type)
+      : ns_id(std::move(ns_id)),
+        table_ids((std::move(table_ids))),
+        record_type(record_type),
+        record_format(record_format),
+        source_type(source_type),
+        checkpoint_type(checkpoint_type) {
   }
 };