yugabyte#4929: Cleanup split tablets

Summary:
Implemented cleanup of the tablet for which all replicas have been split:
- Added DeleteTablet master functionality and RPC support.
- Renamed `TEST_GetAllAppliedOpId` -> `GetAllAppliedOpId` to be used in production code.
- `TSTabletManager`: added tablets cleaner background task which is responsible for deleting of split tablets that are no longer needed.
- Added `TabletSplitITest.ParentTabletCleanup`.

Test Plan:
```
ybd --remote --dltp --cxx-test tablet-split-itest -n 100 -- -p 1
```

Reviewers: hector, nicolas, bogdan

Reviewed By: bogdan

Subscribers: zyu, ybase, bogdan

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

Author: ttyusupov

ent/src/yb/master/catalog_manager-test_ent.cc

@@ -50,7 +50,7 @@ void SetupClusterConfigEnt(const vector<string>& az_list,
                            const vector<string>& affinitized_leader_list,
                            ReplicationInfoPB* replication_info) {
   PlacementInfoPB* placement_info = replication_info->mutable_live_replicas();
-  placement_info->set_num_replicas(kNumReplicas);
+  placement_info->set_num_replicas(kDefaultNumReplicas);
 
   for (const string& az : az_list) {
     auto pb = placement_info->add_placement_blocks();

src/yb/client/client-internal.cc

@@ -1562,6 +1562,45 @@ void GetCDCStreamRpc::ProcessResponse(const Status& status) {
   user_cb_(status);
 }
 
+class DeleteTabletRpc
+    : public ClientMasterRpc<master::DeleteTabletRequestPB, master::DeleteTabletResponsePB> {
+ public:
+  DeleteTabletRpc(
+      YBClient* client,
+      const TabletId& tablet_id,
+      StdStatusCallback user_cb,
+      CoarseTimePoint deadline,
+      rpc::Messenger* messenger,
+      rpc::ProxyCache* proxy_cache)
+      : ClientMasterRpc(client, deadline, messenger, proxy_cache),
+        user_cb_(std::move(user_cb)) {
+    req_.set_tablet_id(tablet_id);
+  }
+
+  std::string ToString() const override {
+    return Format(
+        "DeleteTabletRpc(tablet_id: $0, num_attempts: $1)", req_.tablet_id(), num_attempts());
+  }
+
+  virtual ~DeleteTabletRpc() = default;
+
+ private:
+  void CallRemoteMethod() override {
+    master_proxy()->DeleteTabletAsync(
+        req_, &resp_, mutable_retrier()->mutable_controller(),
+        std::bind(&DeleteTabletRpc::Finished, this, Status::OK()));
+  }
+
+  void ProcessResponse(const Status& status) override {
+    if (!status.ok()) {
+      LOG(WARNING) << ToString() << " failed: " << status.ToString();
+    }
+    user_cb_(status);
+  }
+
+  StdStatusCallback user_cb_;
+};
+
 } // namespace internal
 
 Status YBClient::Data::GetTableSchema(YBClient* client,
@@ -1715,6 +1754,18 @@ void YBClient::Data::GetCDCStream(
       proxy_cache_.get());
 }
 
+void YBClient::Data::DeleteTablet(
+    YBClient* client, const TabletId& tablet_id, CoarseTimePoint deadline,
+    StdStatusCallback callback) {
+  auto rpc = rpc::StartRpc<internal::DeleteTabletRpc>(
+      client,
+      tablet_id,
+      callback,
+      deadline,
+      messenger_,
+      proxy_cache_.get());
+}
+
 void YBClient::Data::LeaderMasterDetermined(const Status& status,
                                             const HostPort& host_port) {
   Status new_status = status;

src/yb/client/client-internal.h

@@ -246,6 +246,10 @@ class YBClient::Data {
                     CoarseTimePoint deadline,
                     StdStatusCallback callback);
 
+  void DeleteTablet(
+      YBClient* client, const TabletId& tablet_id, CoarseTimePoint deadline,
+      StdStatusCallback callback);
+
   CHECKED_STATUS InitLocalHostNames();
 
   bool IsLocalHostPort(const HostPort& hp) const;

src/yb/client/client.cc

@@ -131,6 +131,8 @@ using yb::master::DeleteUDTypeRequestPB;
 using yb::master::DeleteUDTypeResponsePB;
 using yb::master::DeleteRoleRequestPB;
 using yb::master::DeleteRoleResponsePB;
+using yb::master::DeleteTabletRequestPB;
+using yb::master::DeleteTabletResponsePB;
 using yb::master::GetPermissionsRequestPB;
 using yb::master::GetPermissionsResponsePB;
 using yb::master::GrantRevokeRoleRequestPB;
@@ -1354,6 +1356,11 @@ void YBClient::DeleteCDCStream(const CDCStreamId& stream_id, StatusCallback call
   data_->DeleteCDCStream(this, stream_id, deadline, callback);
 }
 
+void YBClient::DeleteTablet(const TabletId& tablet_id, StdStatusCallback callback) {
+  auto deadline = CoarseMonoClock::Now() + default_admin_operation_timeout();
+  data_->DeleteTablet(this, tablet_id, deadline, callback);
+}
+
 Status YBClient::TabletServerCount(int *tserver_count, bool primary_only, bool use_cache) {
   int tserver_count_cached = data_->tserver_count_cached_.load(std::memory_order_acquire);
   if (use_cache && tserver_count_cached > 0) {

src/yb/client/client.h

@@ -521,6 +521,8 @@ class YBClient {
                     std::shared_ptr<std::unordered_map<std::string, std::string>> options,
                     StdStatusCallback callback);
 
+  void DeleteTablet(const TabletId& tablet_id, StdStatusCallback callback);
+
   // Find the number of tservers. This function should not be called frequently for reading or
   // writing actual data. Currently, it is called only for SQL DDL statements.
   // If primary_only is set to true, we expect the primary/sync cluster tserver count only.

src/yb/client/ql-tablet-test.cc

@@ -1239,7 +1239,7 @@ std::vector<OpId> GetLastAppliedOpIds(const std::vector<tablet::TabletPeerPtr>&
 Result<OpId> GetAllAppliedOpId(const std::vector<tablet::TabletPeerPtr>& peers) {
   for (auto& peer : peers) {
     if (peer->LeaderStatus() == consensus::LeaderStatus::LEADER_AND_READY) {
-      return peer->raft_consensus()->TEST_GetAllAppliedOpId();
+      return peer->raft_consensus()->GetAllAppliedOpId();
     }
   }
   return STATUS(NotFound, "No leader found");

src/yb/consensus/consensus_queue-test.cc

@@ -363,7 +363,7 @@ TEST_F(ConsensusQueueTest, TestPeersDontAckBeyondWatermarks) {
   // replicated watermark to the last op appended to the local log.
   ASSERT_EQ(queue_->TEST_GetAllReplicatedIndex(), MakeOpIdForIndex(kNumMessages));
   ASSERT_EQ(queue_->TEST_GetLastAppliedOpId(), OpId::Min());
-  ASSERT_EQ(queue_->TEST_GetAllAppliedOpId(), OpId::Min());
+  ASSERT_EQ(queue_->GetAllAppliedOpId(), OpId::Min());
 
   // Start to track the peer after the queue has some messages in it
   // at a point that is halfway through the current messages in the queue.
@@ -379,7 +379,7 @@ TEST_F(ConsensusQueueTest, TestPeersDontAckBeyondWatermarks) {
   ASSERT_EQ(queue_->TEST_GetAllReplicatedIndex(), OpId::Min());
   ASSERT_EQ(queue_->TEST_GetMajorityReplicatedOpId(), OpId::Min());
   ASSERT_EQ(queue_->TEST_GetLastAppliedOpId(), OpId::Min());
-  ASSERT_EQ(queue_->TEST_GetAllAppliedOpId(), OpId::Min());
+  ASSERT_EQ(queue_->GetAllAppliedOpId(), OpId::Min());
 
   ReplicateMsgsHolder refs;
   bool needs_remote_bootstrap;
@@ -443,7 +443,7 @@ TEST_F(ConsensusQueueTest, TestQueueAdvancesCommittedIndex) {
   queue_->raft_pool_observers_token_->Wait();
   ASSERT_EQ(queue_->TEST_GetCommittedIndex(), OpId::Min());
   ASSERT_EQ(queue_->TEST_GetLastAppliedOpId(), OpId::Min());
-  ASSERT_EQ(queue_->TEST_GetAllAppliedOpId(), OpId::Min());
+  ASSERT_EQ(queue_->GetAllAppliedOpId(), OpId::Min());
 
   // NOTE: We don't need to get operations from the queue. The queue
   // only cares about what the peer reported as received, not what was sent.
@@ -462,7 +462,7 @@ TEST_F(ConsensusQueueTest, TestQueueAdvancesCommittedIndex) {
   queue_->raft_pool_observers_token_->Wait();
   ASSERT_EQ(queue_->TEST_GetCommittedIndex(), OpId::Min());
   ASSERT_EQ(queue_->TEST_GetLastAppliedOpId(), OpId::Min());
-  ASSERT_EQ(queue_->TEST_GetAllAppliedOpId(), OpId::Min());
+  ASSERT_EQ(queue_->GetAllAppliedOpId(), OpId::Min());
 
   // Ack the first five operations for peer-2
   response.set_responder_uuid("peer-2");
@@ -640,7 +640,7 @@ TEST_F(ConsensusQueueTest, TestQueueHandlesOperationOverwriting) {
   // 0.0 since we haven't had a successful exchange with the 'remote' peer.
   ASSERT_EQ(queue_->TEST_GetAllReplicatedIndex(), OpId::Min());
   ASSERT_EQ(queue_->TEST_GetLastAppliedOpId(), committed_op_id);
-  ASSERT_EQ(queue_->TEST_GetAllAppliedOpId(), OpId::Min());
+  ASSERT_EQ(queue_->GetAllAppliedOpId(), OpId::Min());
 
   // Test even when a correct peer responds (meaning we actually get to execute
   // watermark advancement) we sill have the same all-replicated watermark.
@@ -650,7 +650,7 @@ TEST_F(ConsensusQueueTest, TestQueueHandlesOperationOverwriting) {
 
   ASSERT_EQ(queue_->TEST_GetAllReplicatedIndex(), OpId::Min());
   ASSERT_EQ(queue_->TEST_GetLastAppliedOpId(), committed_op_id);
-  ASSERT_EQ(queue_->TEST_GetAllAppliedOpId(), OpId::Min());
+  ASSERT_EQ(queue_->GetAllAppliedOpId(), OpId::Min());
 
   // Generate another request for the remote peer, which should include
   // all of the ops since the peer's last-known committed index.

src/yb/consensus/consensus_queue.cc

@@ -1228,7 +1228,7 @@ OpId PeerMessageQueue::TEST_GetAllReplicatedIndex() const {
   return queue_state_.all_replicated_op_id;
 }
 
-OpId PeerMessageQueue::TEST_GetAllAppliedOpId() const {
+OpId PeerMessageQueue::GetAllAppliedOpId() const {
   LockGuard lock(queue_lock_);
   return queue_state_.all_applied_op_id;
 }

src/yb/consensus/consensus_queue.h

@@ -303,7 +303,7 @@ class PeerMessageQueue {
 
   OpId TEST_GetCommittedIndex() const;
 
-  OpId TEST_GetAllAppliedOpId() const;
+  OpId GetAllAppliedOpId() const;
 
   // Returns the current majority replicated OpId, for tests.
   OpId TEST_GetMajorityReplicatedOpId() const;
@@ -500,7 +500,7 @@ class PeerMessageQueue {
   // Updates op id replicated on each node.
   void UpdateAllReplicatedOpId(OpId* result) REQUIRES(queue_lock_);
 
-  // Updates op id applied on each node.
+  // Updates op ID applied on each node.
   void UpdateAllAppliedOpId(OpId* result) REQUIRES(queue_lock_);
 
   // Updates op id replicated on each non-lagging node.

src/yb/consensus/raft_consensus.cc

@@ -3020,8 +3020,8 @@ yb::OpId RaftConsensus::GetLastAppliedOpId() {
   return state_->GetLastAppliedOpIdUnlocked();
 }
 
-yb::OpId RaftConsensus::TEST_GetAllAppliedOpId() {
-  return queue_->TEST_GetAllAppliedOpId();
+yb::OpId RaftConsensus::GetAllAppliedOpId() {
+  return queue_->GetAllAppliedOpId();
 }
 
 yb::OpId RaftConsensus::GetSplitOpId() {