Skip to content

Commit 1b50b15

Browse files
committed
[BACKPORT 2.12] [CDCSDK] [#9019] CDC SDK API changes
Summary: Original commits: - b04b58a / D14605 - 9ac94af / D14990 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 ``` [#11205] Trivial RPC methods Currently, most of our RPC endpoints has a trivial implementation, which takes request, synchronously process, and sends the response. But our generated service methods allow asynchronous processing, which complicates the writing of every method, even it doesn't require asynchronous processing. This diff adds the ability to specify that method is trivial. So his service virtual method will have signature: ```Result<ResponsePB> Method(const RequestPB& req, CoarseTimePoint deadline);``` instead of ```void Method(const RequestPB req, ResponsePB* resp, RpcContext context);``` Using std instead of boost lib for declval Test Plan: Jenkins: skip Added unit tests. Have tested it manually. ybd --gtest_filter RpcStubTest.Trivial Reviewers: rahuldesirazu, nicolas, vkushwaha, dmitry, sergei Reviewed By: sergei Subscribers: bogdan Differential Revision: https://phabricator.dev.yugabyte.com/D16228
1 parent aef685b commit 1b50b15

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

46 files changed

+2629
-631
lines changed

build-support/common-build-env.sh

+1
Original file line numberDiff line numberDiff line change
@@ -2211,6 +2211,7 @@ activate_virtualenv() {
22112211
pip_no_cache="--no-cache-dir"
22122212
fi
22132213

2214+
which pip3
22142215
local pip_executable=pip3
22152216
if ! "$yb_readonly_virtualenv"; then
22162217
local requirements_file_path="$YB_SRC_ROOT/requirements_frozen.txt"

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

+2-1
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,8 @@ set(CDC_SRCS_EXTENSIONS
2525
${YB_ENT_CURRENT_SOURCE_DIR}/cdc_service.cc
2626
${YB_ENT_CURRENT_SOURCE_DIR}/cdc_metrics.cc
2727
${YB_ENT_CURRENT_SOURCE_DIR}/cdc_producer.cc
28-
${YB_ENT_CURRENT_SOURCE_DIR}/cdc_rpc.cc)
28+
${YB_ENT_CURRENT_SOURCE_DIR}/cdc_rpc.cc
29+
${YB_ENT_CURRENT_SOURCE_DIR}/cdc_error.cc)
2930

3031
ADD_YB_LIBRARY(
3132
cdc

ent/src/yb/cdc/cdc_error.cc

+43
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
//
18+
// The following only applies to changes made to this file as part of YugaByte development.
19+
//
20+
// Portions Copyright (c) YugaByte, Inc.
21+
//
22+
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
23+
// in compliance with the License. You may obtain a copy of the License at
24+
//
25+
// http://www.apache.org/licenses/LICENSE-2.0
26+
//
27+
// Unless required by applicable law or agreed to in writing, software distributed under the License
28+
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
29+
// or implied. See the License for the specific language governing permissions and limitations
30+
// under the License.
31+
//
32+
#include "yb/cdc/cdc_error.h"
33+
34+
namespace yb {
35+
namespace cdc {
36+
37+
static const std::string kCDCErrorCategoryName = "CDC error";
38+
39+
static StatusCategoryRegisterer cdc_error_category_registerer(
40+
StatusCategoryDescription::Make<CDCErrorTag>(&kCDCErrorCategoryName));
41+
42+
} // namespace cdc
43+
} // namespace yb

ent/src/yb/cdc/cdc_error.h

+59
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
//
18+
// The following only applies to changes made to this file as part of YugaByte development.
19+
//
20+
// Portions Copyright (c) YugaByte, Inc.
21+
//
22+
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
23+
// in compliance with the License. You may obtain a copy of the License at
24+
//
25+
// http://www.apache.org/licenses/LICENSE-2.0
26+
//
27+
// Unless required by applicable law or agreed to in writing, software distributed under the License
28+
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
29+
// or implied. See the License for the specific language governing permissions and limitations
30+
// under the License.
31+
//
32+
33+
#ifndef ENT_SRC_YB_CDC_CDC_ERROR_H
34+
#define ENT_SRC_YB_CDC_CDC_ERROR_H
35+
36+
#include "yb/cdc/cdc_service.pb.h"
37+
38+
#include "yb/util/status_ec.h"
39+
40+
namespace yb {
41+
namespace cdc {
42+
43+
struct CDCErrorTag : IntegralErrorTag<CDCErrorPB::Code> {
44+
// This category id is part of the wire protocol and should not be changed once released.
45+
static constexpr uint8_t kCategory = 16;
46+
47+
typedef CDCErrorPB::Code Code;
48+
49+
static const std::string& ToMessage(Code code) {
50+
return CDCErrorPB::Code_Name(code);
51+
}
52+
};
53+
54+
typedef StatusErrorCodeImpl<CDCErrorTag> CDCError;
55+
56+
} // namespace cdc
57+
} // namespace yb
58+
59+
#endif // ENT_SRC_YB_CDC_CDC_ERROR_H

ent/src/yb/cdc/cdc_producer.h

+17-3
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020
#include <boost/unordered_map.hpp>
2121

2222
#include "yb/cdc/cdc_service.service.h"
23+
#include "yb/common/common_fwd.h"
2324
#include "yb/common/transaction.h"
2425
#include "yb/consensus/consensus_fwd.h"
2526
#include "yb/docdb/docdb.pb.h"
@@ -34,14 +35,27 @@ class MemTracker;
3435
namespace cdc {
3536

3637
struct StreamMetadata {
37-
TableId table_id;
38+
NamespaceId ns_id;
39+
std::vector<TableId> table_ids;
3840
CDCRecordType record_type;
3941
CDCRecordFormat record_format;
42+
CDCRequestSource source_type;
43+
CDCCheckpointType checkpoint_type;
4044

4145
StreamMetadata() = default;
4246

43-
StreamMetadata(TableId table_id, CDCRecordType record_type, CDCRecordFormat record_format)
44-
: table_id(std::move(table_id)), record_type(record_type), record_format(record_format) {
47+
StreamMetadata(NamespaceId ns_id,
48+
std::vector<TableId> table_ids,
49+
CDCRecordType record_type,
50+
CDCRecordFormat record_format,
51+
CDCRequestSource source_type,
52+
CDCCheckpointType checkpoint_type)
53+
: ns_id(std::move(ns_id)),
54+
table_ids((std::move(table_ids))),
55+
record_type(record_type),
56+
record_format(record_format),
57+
source_type(source_type),
58+
checkpoint_type(checkpoint_type) {
4559
}
4660
};
4761

0 commit comments

Comments
 (0)