[yugabyte#8485] Refactor split op info to generic operation blocking mechanism

Summary:
This diff removes split op id from replica state. This split op id was being used for the various purposes:
1) Block adding new write (and some other) operations to the Raft log after the split operation started to replicate.
2) Store data for child tablet ids.
3) Retain split op id in logs.

It was changed to the following:
1) Introduced a generic mechanism to dynamically reject operations.
2) Store split op id and child tablet ids in the tablet metadata (superblock). This is a persistent format change.
3) Use log anchor registry.

Also moved lease check higher by call stack and do it only for the last operation in batch.

Removed an unused index of a multi index container (on the op id) from `RunningRetryableRequests`.

Changes to duplicate request error handling
===========================================

RetryableRequests is a subsystem we use for filtering out duplicate requests that could have been submitted during retries. All operations submitted to Raft are also registered with RetryableRequests. For a duplicate requests, we handle failure as follows. Either the Register function notifies the duplicate operation that it failed immediately (e.g. in case the original operation is known to have failed), or it waits for the original request to finish replicating and then notifies the duplicate operation of replication failure. After that the operation is removed from MvccManager and is rolled back in ReplicaState using its op id.

Also, when we add operations to Raft consensus, we do it in batches, and if at least one operation fails to get added, we fail the entire batch.

There are two issues with the interaction between these mechanisms.
1. A duplicate request might be waiting for the original operation to finish replicating, and therefore the duplicate request is not deleted from MvccManager right away. As a result, an operation with the same op id could be added to MvccManager. Prior to the recent commit 1aece37 (D11371), this was not an issue, because MvccManager only contained hybrid time (but no op ids) and the duplicate operation's hybrid time was greater than the original operation's hybrid time, so the read path was not affected (and we allow handling aborts in any order).
2. If the operation that fails is not the first operation in its batch (which would mean we've already called RetryableRequests::Register on some earlier operations in the batch), we call ReplicationFinished for all preceding operations in the batch, and no one would notify RetryableRequests about the completion of those preceding operations, so they would stay in RetryableRequests indefinitely.

The above issues became more frequent with this diff because we now check the lease after adding all operations. In the most frequent case when we have a valid leader lease, with the new behavior we will spend less time checking lease, and would not be reading the system clock repeatedly. However, the new logic might be slightly suboptimal in the rare case when the lease has expired.

The second scenario above has become more frequent with the changes in this diff (but before the fix described below) because with the lease check happening after adding all operations in the batch, we would have added all the operations to RetryableRequests. Prior to this diff, for operations to get stuck in RetryableRequests in this way, the lease check would need to succeed e.g. for the first operation so it would get added to RetryableRequests, but then it would need to fail for the second operation, and this combination of circumstances is very rare.

To fix the above issue, in this diff we handle 3 different cases:

1. Rounds that were rejected by RetryableRequests. We should not call ReplicationFinished for them as it would have been already called by Register.
2. Rounds that were registered with RetryableRequests. We should call NotifyReplicationFinishedUnlocked for them, which also notifies RetryableRequests of the operation failure.
3. Rounds that were not registered with retryable requests. We should call ReplicationFinished directly for them.

Test Plan: Jenkins

Reviewers: timur, mbautin

Reviewed By: mbautin

Subscribers: mbautin, bogdan, ybase

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

Author: spolitov

src/yb/consensus/consensus.cc

@@ -128,14 +128,12 @@ Status Consensus::ExecuteHook(HookPoint point) {
   return Status::OK();
 }
 
-Result<yb::OpId> Consensus::GetLastOpId(OpIdType type) {
+Result<OpId> Consensus::GetLastOpId(OpIdType type) {
   switch (type) {
     case OpIdType::RECEIVED_OPID:
       return GetLastReceivedOpId();
     case OpIdType::COMMITTED_OPID:
       return GetLastCommittedOpId();
-    case OpIdType::SPLIT_OPID:
-      return GetSplitOpId();
     case OpIdType::UNKNOWN_OPID_TYPE:
       break;
   }

src/yb/consensus/consensus.h

@@ -62,10 +62,6 @@
 
 namespace yb {
 
-namespace log {
-class Log;
-}
-
 namespace master {
 class SysCatalogTable;
 }
@@ -101,12 +97,6 @@ struct ConsensusBootstrapInfo {
   // The id of the last committed operation in the log.
   OpIdPB last_committed_id;
 
-  // Parameters of the split operation added to Raft log and designated for this tablet (in general
-  // case Raft log the tablet could contain old split operations designated for it ancestor tablets,
-  // those are not reflected here).
-  // See comments for ReplicateState::split_op_with_tablet_ids_.
-  SplitOpInfo split_op_info;
-
   // REPLICATE messages which were in the log with no accompanying
   // COMMIT. These need to be passed along to consensus init in order
   // to potentially commit them.
@@ -240,7 +230,7 @@ class Consensus {
   virtual CHECKED_STATUS TEST_Replicate(const ConsensusRoundPtr& round) = 0;
 
   // A batch version of Replicate, which is what we try to use as much as possible for performance.
-  virtual CHECKED_STATUS ReplicateBatch(ConsensusRounds* rounds) = 0;
+  virtual CHECKED_STATUS ReplicateBatch(const ConsensusRounds& rounds) = 0;
 
   // Messages sent from LEADER to FOLLOWERS and LEARNERS to update their
   // state machines. This is equivalent to "AppendEntries()" in Raft
@@ -336,14 +326,6 @@ class Consensus {
 
   virtual yb::OpId GetLastAppliedOpId() = 0;
 
-  // Return the ID of the split operation requesting to split this Raft group if it has been added
-  // to Raft log and uninitialized OpId otherwise.
-  virtual yb::OpId GetSplitOpId() = 0;
-
-  // Return split child tablet IDs if split operation has been added to Raft log and array of empty
-  // tablet IDs otherwise.
-  virtual std::array<TabletId, kNumSplitParts> GetSplitChildTabletIds() = 0;
-
   // Assuming we are the leader, wait until we have a valid leader lease (i.e. the old leader's
   // lease has expired, and we have replicated a new lease that has not expired yet).
   virtual CHECKED_STATUS WaitForLeaderLeaseImprecise(CoarseTimePoint deadline) = 0;
@@ -520,8 +502,8 @@ class ConsensusRound : public RefCountedThreadSafe<ConsensusRound> {
 
   // Returns the id of the (replicate) operation this context
   // refers to. This is only set _after_ Consensus::Replicate(context).
-  OpIdPB id() const {
-    return replicate_msg_->id();
+  OpId id() const {
+    return OpId::FromPB(replicate_msg_->id());
   }
 
   // Register a callback that is called by Consensus to notify that the round
@@ -549,7 +531,6 @@ class ConsensusRound : public RefCountedThreadSafe<ConsensusRound> {
   // other than 'term'.
   // See CheckBoundTerm().
   void BindToTerm(int64_t term) {
-    DCHECK_EQ(bound_term_, kUnboundTerm);
     bound_term_ = term;
   }
 

src/yb/consensus/consensus.proto

@@ -542,7 +542,6 @@ enum OpIdType {
   UNKNOWN_OPID_TYPE = 0;
   RECEIVED_OPID = 1;
   COMMITTED_OPID = 2;
-  SPLIT_OPID = 3;
 }
 
 message GetLastOpIdRequestPB {
@@ -554,6 +553,8 @@ message GetLastOpIdRequestPB {
 
   // Whether to return the last-received or last-committed OpId.
   optional OpIdType opid_type = 3 [ default = RECEIVED_OPID ];
+
+  optional OperationType op_type = 4;
 }
 
 message GetLastOpIdResponsePB {

src/yb/consensus/consensus_context.h

@@ -17,6 +17,7 @@
 #include "yb/common/common_fwd.h"
 
 #include "yb/consensus/consensus_fwd.h"
+#include "yb/consensus/consensus.pb.h"
 
 #include "yb/util/result.h"
 
@@ -69,6 +70,10 @@ class ConsensusContext {
   // Listener could be set only once and then reset.
   virtual void ListenNumSSTFilesChanged(std::function<void()> listener) = 0;
 
+  // Checks whether operation with provided op id and type could be added to the log.
+  virtual CHECKED_STATUS CheckOperationAllowed(
+      const OpId& op_id, consensus::OperationType op_type) = 0;
+
   virtual ~ConsensusContext() = default;
 };
 

src/yb/consensus/consensus_fwd.h

@@ -19,13 +19,6 @@
 #include "yb/util/strongly_typed_bool.h"
 
 namespace yb {
-
-namespace log {
-
-class Log;
-
-} // namespace log
-
 namespace consensus {
 
 class Consensus;

src/yb/consensus/consensus_peers.h

@@ -62,10 +62,6 @@ class Messenger;
 class PeriodicTimer;
 }
 
-namespace log {
-class Log;
-}
-
 namespace consensus {
 
 // A peer in consensus (local or remote).

src/yb/consensus/consensus_queue.cc

@@ -147,7 +147,6 @@ DEFINE_validator(rpc_throttle_threshold_bytes, &RpcThrottleThresholdBytesValidat
 namespace yb {
 namespace consensus {
 
-using log::AsyncLogReader;
 using log::Log;
 using std::unique_ptr;
 using rpc::Messenger;
@@ -1604,5 +1603,10 @@ void PeerMessageQueue::TrackOperationsMemory(const OpIds& op_ids) {
   log_cache_.TrackOperationsMemory(op_ids);
 }
 
+Result<OpId> PeerMessageQueue::TEST_GetLastOpIdWithType(
+    int64_t max_allowed_index, OperationType op_type) {
+  return log_cache_.TEST_GetLastOpIdWithType(max_allowed_index, op_type);
+}
+
 }  // namespace consensus
 }  // namespace yb

src/yb/consensus/consensus_queue.h

@@ -63,11 +63,6 @@ class MemTracker;
 class MetricEntity;
 class ThreadPoolToken;
 
-namespace log {
-class Log;
-class AsyncLogReader;
-}
-
 namespace consensus {
 class PeerMessageQueueObserver;
 struct MajorityReplicatedData;
@@ -378,6 +373,8 @@ class PeerMessageQueue {
     return clock_;
   }
 
+  Result<OpId> TEST_GetLastOpIdWithType(int64_t max_allowed_index, OperationType op_type);
+
  private:
   FRIEND_TEST(ConsensusQueueTest, TestQueueAdvancesCommittedIndex);
   FRIEND_TEST(ConsensusQueueTest, TestReadReplicatedMessagesForCDC);

src/yb/consensus/consensus_types.h

@@ -36,19 +36,19 @@ class ConsensusAppendCallback {
   // committed_op_id - committed operation id.
   //
   // Should initialize appropriate replicate message.
-  virtual void HandleConsensusAppend(const yb::OpId& op_id, const yb::OpId& committed_op_id) = 0;
+  virtual void HandleConsensusAppend(const OpId& op_id, const OpId& committed_op_id) = 0;
+
+  // Invoked when appropriate operation failed to replicate.
+  virtual void ReplicationFinished(
+      const Status& status, int64_t leader_term, OpIds* applied_op_ids) = 0;
+
   virtual ~ConsensusAppendCallback() {}
 };
 
 struct ConsensusOptions {
   std::string tablet_id;
 };
 
-struct SplitOpInfo {
-  OpId op_id;
-  std::array<TabletId, kNumSplitParts> child_tablet_ids;
-};
-
 } // namespace consensus
 } // namespace yb
 

src/yb/consensus/log.cc

@@ -642,7 +642,7 @@ Status Log::RollOver() {
   return Status::OK();
 }
 
-Status Log::Reserve(LogEntryTypePB type,
+void Log::Reserve(LogEntryTypePB type,
                     LogEntryBatchPB* entry_batch,
                     LogEntryBatch** reserved_entry) {
   TRACE_EVENT0("log", "Log::Reserve");
@@ -669,7 +669,6 @@ Status Log::Reserve(LogEntryTypePB type,
   // TODO (perf) Use a ring buffer instead of a blocking queue and set
   // 'reserved_entry' to a pre-allocated slot in the buffer.
   *reserved_entry = new_entry_batch.release();
-  return Status::OK();
 }
 
 Status Log::TEST_AsyncAppendWithReplicates(
@@ -715,7 +714,7 @@ Status Log::AsyncAppendReplicates(const ReplicateMsgs& msgs, const yb::OpId& com
   }
 
   LogEntryBatch* reserved_entry_batch;
-  RETURN_NOT_OK(Reserve(REPLICATE, &batch, &reserved_entry_batch));
+  Reserve(REPLICATE, &batch, &reserved_entry_batch);
 
   // If we're able to reserve, set the vector of replicate shared pointers in the LogEntryBatch.
   // This will make sure there's a reference for each replicate while we're appending.
@@ -1052,7 +1051,7 @@ Status Log::WaitUntilAllFlushed() {
   LogEntryBatchPB entry_batch;
   entry_batch.add_entry()->set_type(log::FLUSH_MARKER);
   LogEntryBatch* reserved_entry_batch;
-  RETURN_NOT_OK(Reserve(FLUSH_MARKER, &entry_batch, &reserved_entry_batch));
+  Reserve(FLUSH_MARKER, &entry_batch, &reserved_entry_batch);
   Synchronizer s;
   RETURN_NOT_OK(AsyncAppend(reserved_entry_batch, s.AsStatusCallback()));
   return s.Wait();