yugabyte#2050: Increase WAL retention time when CDC stream is setup o…

…n table

Summary: When a CDC stream is created, we modify the WAL retention time for all the table's tablets. This is done with an AlterTable request. The new time is controlled by the gflag cdc_wal_retention_time_secs, and it is the same for any table for which a CDC stream is created.

Test Plan: Verify that all the tests that create a CDC stream succeed

Reviewers: bogdan, rahuldesirazu, neha

Reviewed By: neha

Subscribers: ybase

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

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

@@ -41,5 +41,6 @@ target_link_libraries(cdc
   tablet
   yb_util
   ql_util
-  gutil)
+  gutil
+  yb_client)
 

ent/src/yb/cdc/cdc_service.cc

@@ -22,6 +22,7 @@
 #include "yb/common/wire_protocol.h"
 #include "yb/consensus/raft_consensus.h"
 #include "yb/client/table.h"
+#include "yb/client/table_alterer.h"
 #include "yb/client/table_handle.h"
 #include "yb/client/session.h"
 #include "yb/client/yb_table_name.h"
@@ -48,6 +49,10 @@ TAG_FLAG(cdc_ybclient_reactor_threads, advanced);
 DEFINE_test_flag(bool, mock_get_changes_response_for_consumer_testing, false,
                  "Mock a successful response to consumer before stream id integration is set up.");
 
+DEFINE_int32(cdc_wal_retention_time_secs, 4 * 3600,
+             "WAL retention time in seconds to be used for tables for which a CDC stream was "
+             "created.");
+
 namespace yb {
 namespace cdc {
 
@@ -155,26 +160,38 @@ void CDCServiceImpl::CreateCDCStream(const CreateCDCStreamRequestPB* req,
     // Check if YSQL table has a primary key. CQL tables always have a user specified primary key.
     RPC_CHECK_AND_RETURN_ERROR(
         table->table_type() != client::YBTableType::PGSQL_TABLE_TYPE ||
-          YsqlTableHasPrimaryKey(table->schema()),
+        YsqlTableHasPrimaryKey(table->schema()),
         STATUS(InvalidArgument, "Cannot setup CDC on table without primary key"),
         resp->mutable_error(),
         CDCErrorPB::INVALID_REQUEST,
         context);
   }
 
+  std::unique_ptr<client::YBTableAlterer> table_alterer(
+      async_client_init_->client()->NewTableAlterer(table->name()));
+
+  // Increase WAL retention. Fail the request if we fail to set the WAL retention.
+  auto wal_retention_secs = 0;
+  if (req->has_retention_sec()) {
+    wal_retention_secs = req->retention_sec();
+  } else {
+    wal_retention_secs = static_cast<uint32>(FLAGS_cdc_wal_retention_time_secs);
+  }
+
+  auto status = table_alterer->SetWalRetentionSecs(wal_retention_secs)->Alter();
+  RPC_STATUS_RETURN_ERROR(status, resp->mutable_error(), CDCErrorPB::INTERNAL_ERROR, context);
+
   std::unordered_map<std::string, std::string> options;
   options.reserve(3);
   options.emplace(kRecordType, CDCRecordType_Name(req->record_type()));
   options.emplace(kRecordFormat, CDCRecordFormat_Name(req->record_format()));
-  if (req->has_retention_sec()) {
-    options.emplace(kRetentionSec, std::to_string(req->retention_sec()));
-  }
+  options.emplace(kRetentionSec, std::to_string(wal_retention_secs));
+
   auto result = async_client_init_->client()->CreateCDCStream(req->table_id(), options);
   RPC_CHECK_AND_RETURN_ERROR(result.ok(), result.status(), resp->mutable_error(),
                              CDCErrorPB::INTERNAL_ERROR, context);
 
   resp->set_stream_id(*result);
-  // TODO: Increase retention for WAL.
 
   // Add stream to cache.
   AddStreamMetadataToCache(*result, std::make_shared<StreamMetadata>(req->table_id(),

ent/src/yb/integration-tests/cdc_service-int-test.cc

@@ -24,6 +24,8 @@
 #include "yb/yql/cql/ql/util/errcodes.h"
 #include "yb/yql/cql/ql/util/statement_result.h"
 
+DECLARE_int32(cdc_wal_retention_time_secs);
+
 namespace yb {
 namespace cdc {
 
@@ -73,6 +75,9 @@ class CDCServiceTest : public YBMiniClusterTestBase<MiniCluster> {
   void CreateTable(int num_tablets, TableHandle* table);
   void GetTablet(std::string* tablet_id);
 
+  void VerifyWalRetentionTime(const TableId& table_id,
+                              const uint32_t expected_wal_retention_secs);
+
   std::unique_ptr<CDCServiceProxy> cdc_proxy_;
   std::unique_ptr<client::YBClient> client_;
   TableHandle table_;
@@ -93,6 +98,25 @@ void CDCServiceTest::CreateTable(int num_tablets, TableHandle* table) {
   ASSERT_OK(table->Create(kTableName, num_tablets, client_.get(), &builder));
 }
 
+void CDCServiceTest::VerifyWalRetentionTime(const TableId& table_id,
+                                            const uint32_t expected_wal_retention_secs) {
+  vector<std::shared_ptr<tablet::TabletPeer> > peers;
+  cluster_->mini_tablet_server(0)->server()->tablet_manager()->GetTabletPeers(&peers);
+
+  bool wal_retention_time_checked = false;
+  // Find the right tablet peer.
+  for (const auto& peer : peers) {
+    if (peer->tablet_metadata()->table_id() == table_id) {
+      LOG(INFO) << "Checking wal retention time for tablet " << peer->tablet()->tablet_id();
+      ASSERT_EQ(peer->log()->wal_retention_secs(), expected_wal_retention_secs);
+      ASSERT_EQ(peer->tablet_metadata()->wal_retention_secs(), expected_wal_retention_secs);
+      wal_retention_time_checked = true;
+      break;
+    }
+  }
+  ASSERT_TRUE(wal_retention_time_checked);
+}
+
 void AssertChangeRecords(const google::protobuf::RepeatedPtrField<cdc::KeyValuePairPB>& changes,
                          int32_t expected_int, std::string expected_str) {
   ASSERT_EQ(changes.size(), 2);
@@ -112,17 +136,50 @@ void CDCServiceTest::GetTablet(std::string* tablet_id) {
 
 TEST_F(CDCServiceTest, TestCreateCDCStream) {
   CDCStreamId stream_id;
-  CreateCDCStream(cdc_proxy_, table_.table()->id(), &stream_id);
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* retention time */, &stream_id);
 
   TableId table_id;
   std::unordered_map<std::string, std::string> options;
   ASSERT_OK(client_->GetCDCStream(stream_id, &table_id, &options));
   ASSERT_EQ(table_id, table_.table()->id());
 }
 
+TEST_F(CDCServiceTest, TestCreateCDCStreamWithDefaultRententionTime) {
+  // Set default WAL retention time to 10 hours.
+  FLAGS_cdc_wal_retention_time_secs = 36000;
+
+  CDCStreamId stream_id;
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* retention time */, &stream_id);
+
+  TableId table_id;
+  std::unordered_map<std::string, std::string> options;
+  ASSERT_OK(client_->GetCDCStream(stream_id, &table_id, &options));
+
+  // Verify that the wal retention time was set at the tablet level.
+  VerifyWalRetentionTime(table_id, FLAGS_cdc_wal_retention_time_secs);
+}
+
+TEST_F(CDCServiceTest, TestCreateCDCStreamWithSpecifiedtRententionTime) {
+  // Set default WAL retention time to 10 hours.
+  FLAGS_cdc_wal_retention_time_secs = 36000;
+
+  // Set WAL retention time to 1 hour.
+  constexpr uint32_t wal_retention_secs = 3600;
+
+  CDCStreamId stream_id;
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), wal_retention_secs, &stream_id);
+
+  TableId table_id;
+  std::unordered_map<std::string, std::string> options;
+  ASSERT_OK(client_->GetCDCStream(stream_id, &table_id, &options));
+
+  // Verify that the wal retention time was set at the tablet level.
+  VerifyWalRetentionTime(table_id, wal_retention_secs);
+}
+
 TEST_F(CDCServiceTest, TestDeleteCDCStream) {
   CDCStreamId stream_id;
-  CreateCDCStream(cdc_proxy_, table_.table()->id(), &stream_id);
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* retention time */, &stream_id);
 
   TableId table_id;
   std::unordered_map<std::string, std::string> options;
@@ -140,7 +197,7 @@ TEST_F(CDCServiceTest, TestDeleteCDCStream) {
 
 TEST_F(CDCServiceTest, TestGetChanges) {
   CDCStreamId stream_id;
-  CreateCDCStream(cdc_proxy_, table_.table()->id(), &stream_id);
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* retention time */, &stream_id);
 
   std::string tablet_id;
   GetTablet(&tablet_id);
@@ -260,7 +317,7 @@ TEST_F(CDCServiceTest, TestGetChanges) {
 
 TEST_F(CDCServiceTest, TestGetCheckpoint) {
   CDCStreamId stream_id;
-  CreateCDCStream(cdc_proxy_, table_.table()->id(), &stream_id);
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* retention time */, &stream_id);
 
   std::string tablet_id;
   GetTablet(&tablet_id);
@@ -284,7 +341,7 @@ TEST_F(CDCServiceTest, TestGetCheckpoint) {
 
 TEST_F(CDCServiceTest, TestListTablets) {
   CDCStreamId stream_id;
-  CreateCDCStream(cdc_proxy_, table_.table()->id(), &stream_id);
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* retention time */, &stream_id);
 
   std::string tablet_id;
   GetTablet(&tablet_id);
@@ -308,7 +365,7 @@ TEST_F(CDCServiceTest, TestListTablets) {
 
 TEST_F(CDCServiceTest, TestOnlyGetLocalChanges) {
   CDCStreamId stream_id;
-  CreateCDCStream(cdc_proxy_, table_.table()->id(), &stream_id);
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* retention time */, &stream_id);
 
   std::string tablet_id;
   GetTablet(&tablet_id);

ent/src/yb/integration-tests/cdc_service-txn-test.cc

@@ -102,7 +102,8 @@ TEST_F(CDCServiceTxnTest, TestGetChanges) {
 
   // Create CDC stream on table.
   CDCStreamId stream_id;
-  CreateCDCStream(cdc_proxy_, table_.table()->id(), &stream_id);
+  CreateCDCStream(cdc_proxy_, table_.table()->id(), boost::none /* wal retention time */,
+      &stream_id);
 
   GetChangesRequestPB change_req;
   GetChangesResponsePB change_resp;

ent/src/yb/util/cdc_test_util.cc

@@ -27,10 +27,14 @@ void AssertIntKey(const google::protobuf::RepeatedPtrField<cdc::KeyValuePairPB>&
 
 void CreateCDCStream(const std::unique_ptr<CDCServiceProxy>& cdc_proxy,
                      const TableId& table_id,
+                     boost::optional<uint32_t> wal_retention_secs,
                      CDCStreamId* stream_id) {
   CreateCDCStreamRequestPB req;
   CreateCDCStreamResponsePB resp;
   req.set_table_id(table_id);
+  if (wal_retention_secs) {
+    req.set_retention_sec(*wal_retention_secs);
+  }
 
   rpc::RpcController rpc;
   cdc_proxy->CreateCDCStream(req, &resp, &rpc);

ent/src/yb/util/cdc_test_util.h

@@ -27,6 +27,7 @@ void AssertIntKey(const google::protobuf::RepeatedPtrField<cdc::KeyValuePairPB>&
 
 void CreateCDCStream(const std::unique_ptr<CDCServiceProxy>& cdc_proxy,
                      const TableId& table_id,
+                     boost::optional<uint32_t> wal_retention_secs,
                      CDCStreamId* stream_id);
 
 } // namespace cdc

src/yb/consensus/log.cc

@@ -786,6 +786,7 @@ Status Log::WaitUntilAllFlushed() {
 }
 
 void Log::set_wal_retention_secs(uint32_t wal_retention_secs) {
+  LOG(INFO) << "Setting wal retention time to " << wal_retention_secs << " seconds";
   wal_retention_secs_.store(wal_retention_secs, std::memory_order_release);
 }