[yugabyte#8254] No leader lease needed for BackfillIndex

Summary:
When backfill index was designed, the master was supposed to spawn out the
backfill tasks to the tservers. This was done by selecting the leader tablet for
each tablet. However, it is not required that the backfill be done at the leader.

Any of the followers that is past the backfilling_time can perform the backfill task;

This diff allows followers (i.e. non-leaders) to also perform backfill. While the design
is still to perform the backfill at the leader, this change is useful to ensure that backfill
is not stuck/gets timed-out in case there is a leadership change.

Test Plan: ybd --cxx-test integration-tests_cassandra_cpp_driver-test --gtest_filter CppCassandraDriverTest.TestTableBackfillWithLeaderMoves -n 10 --tp 1

Reviewers: bogdan, jason

Reviewed By: jason

Subscribers: ybase

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

Author: amitanandaiyer

src/yb/integration-tests/cassandra_cpp_driver-test.cc

@@ -266,6 +266,49 @@ class CppCassandraDriverTestIndexMultipleChunks : public CppCassandraDriverTestI
   }
 };
 
+class CppCassandraDriverTestIndexMultipleChunksWithLeaderMoves
+    : public CppCassandraDriverTestIndexMultipleChunks {
+ public:
+  std::vector<std::string> ExtraMasterFlags() override {
+    auto flags = CppCassandraDriverTestIndex::ExtraMasterFlags();
+    flags.push_back("--enable_load_balancing=true");
+    flags.push_back("--index_backfill_rpc_max_retries=0");
+    // We do not want backfill to fail because of any throttling.
+    flags.push_back("--index_backfill_rpc_timeout_ms=180000");
+    return flags;
+  }
+
+  std::vector<std::string> ExtraTServerFlags() override {
+    auto flags = CppCassandraDriverTestIndex::ExtraTServerFlags();
+    flags.push_back("--backfill_index_rate_rows_per_sec=10");
+    flags.push_back("--backfill_index_write_batch_size=2");
+    return flags;
+  }
+
+  void SetUp() override {
+    CppCassandraDriverTestIndex::SetUp();
+    thread_holder_.AddThreadFunctor([this] {
+      const auto kNumTServers = cluster_->num_tablet_servers();
+      constexpr auto kSleepTimeMs = 5000;
+      for (int i = 0; !thread_holder_.stop_flag(); i++) {
+        const auto tserver_id = i % kNumTServers;
+        ASSERT_OK(cluster_->AddTServerToLeaderBlacklist(
+            cluster_->master(), cluster_->tablet_server(tserver_id)));
+        SleepFor(MonoDelta::FromMilliseconds(kSleepTimeMs));
+        ASSERT_OK(cluster_->EmptyBlacklist(cluster_->master()));
+      }
+    });
+  }
+
+  void TearDown() override {
+    thread_holder_.Stop();
+    CppCassandraDriverTestIndex::TearDown();
+  }
+
+ private:
+  TestThreadHolder thread_holder_;
+};
+
 class CppCassandraDriverTestIndexSlowBackfill : public CppCassandraDriverTestIndex {
  public:
   std::vector<std::string> ExtraMasterFlags() override {
@@ -1862,6 +1905,13 @@ TEST_F_EX(CppCassandraDriverTest, TestTableBackfillInChunks,
                          IncludeAllColumns::kTrue, UserEnforced::kFalse);
 }
 
+TEST_F_EX(
+    CppCassandraDriverTest, TestTableBackfillWithLeaderMoves,
+    CppCassandraDriverTestIndexMultipleChunksWithLeaderMoves) {
+  TestBackfillIndexTable(
+      this, PKOnlyIndex::kFalse, IsUnique::kFalse, IncludeAllColumns::kTrue, UserEnforced::kFalse);
+}
+
 TEST_F_EX(CppCassandraDriverTest, TestTableBackfillUniqueInChunks,
           CppCassandraDriverTestIndexMultipleChunks) {
   TestBackfillIndexTable(this, PKOnlyIndex::kFalse, IsUnique::kTrue,

src/yb/master/async_rpc_tasks.cc

@@ -321,7 +321,7 @@ bool RetryingTSRpcTask::RescheduleWithBackoffDelay() {
   }
 
   LOG_WITH_PREFIX(INFO) << "Scheduling retry with a delay of " << delay_millis
-                        << "ms (attempt = " << attempt_ << ")...";
+                        << "ms (attempt = " << attempt_ << " / " << attempt_threshold << ")...";
 
   if (!PerformStateTransition(task_state, MonitoredTaskState::kScheduling)) {
     LOG_WITH_PREFIX(WARNING) << "Unable to mark this task as MonitoredTaskState::kScheduling";

src/yb/master/backfill_index.cc

@@ -1370,9 +1370,14 @@ void BackfillChunk::UnregisterAsyncTaskCallback() {
 
   if (resp_.has_error()) {
     status = StatusFromPB(resp_.error().status());
-    for (int i = 0; i < resp_.failed_index_ids_size(); i++) {
-      VLOG(1) << " Added to failed index " << resp_.failed_index_ids(i);
-      failed_indexes.insert(resp_.failed_index_ids(i));
+    if (resp_.failed_index_ids_size() > 0) {
+      for (int i = 0; i < resp_.failed_index_ids_size(); i++) {
+        VLOG(1) << " Added to failed index " << resp_.failed_index_ids(i);
+        failed_indexes.insert(resp_.failed_index_ids(i));
+      }
+    } else {
+      // No specific index was marked as a failure. So consider all of them as failed.
+      failed_indexes = indexes_being_backfilled_;
     }
   } else if (state() != MonitoredTaskState::kComplete) {
     // There is no response, so the error happened even before we could

src/yb/tserver/tablet_service.cc

@@ -684,9 +684,12 @@ void TabletServiceAdminImpl::BackfillIndex(
 
   // Wait for SafeTime to get past read_at;
   const HybridTime read_at(req->read_at_hybrid_time());
-  const auto safe_time = tablet.peer->tablet()->SafeTime(
-      tablet::RequireLease::kTrue, read_at, deadline);
+  DVLOG(1) << "Waiting for safe time to be past " << read_at;
+  const auto safe_time =
+      tablet.peer->tablet()->SafeTime(tablet::RequireLease::kFalse, read_at, deadline);
+  DVLOG(1) << "Got safe time " << safe_time.ToString();
   if (!safe_time.ok()) {
+    LOG(ERROR) << "Could not get a good enough safe time " << safe_time.ToString();
     SetupErrorAndRespond(
         resp->mutable_error(),
         safe_time.status(),