[#1258]: Send election request from master during table creation

Summary:
For each tablet, the master selects a replica as a proposed leader on creation. This selection takes into account affinitized leaders as well as leader blacklisting.
When the master receives a heartbeat from sufficient replicas, a `RunLeaderElection` RPC is sent to the proposed leader.

New flag -
    yb-master: `use_create_table_leader_hint` (default = true)

Test Plan:
ybd debug --gtest_filter RaftConsensusITest.TestHintedLeader -n 100

ybd debug --java-test org.yb.cql.TestInsertJson#ttl -n 100

Reviewers: rahuldesirazu, bogdan

Reviewed By: rahuldesirazu, bogdan

Subscribers: zyu, alex, ybase

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

ent/src/yb/master/ts_descriptor.cc

@@ -24,7 +24,7 @@ bool TSDescriptor::IsReadOnlyTS(const ReplicationInfoPB& replication_info) const
   if (placement_info.has_placement_uuid()) {
     return placement_info.placement_uuid() != placement_uuid();
   }
-  return placement_uuid() != "";
+  return !placement_uuid().empty();
 }
 
 bool TSDescriptor::IsAcceptingLeaderLoad(const ReplicationInfoPB& replication_info) const {

java/yb-cql/src/test/java/org/yb/cql/TestTableTTL.java

@@ -363,7 +363,7 @@ public void testRowTTLAfterAlter() throws Exception {
     LOG.info("Insert rows.");
     session.execute(String.format("INSERT INTO %s (c1, c2, c3) values (1, 2, 3);", tableName));
     session.execute(String.format("INSERT INTO %s (c1, c2, c3) values (4, 5, 6) " +
-          "USING TTL 10;", tableName));
+          "USING TTL 6;", tableName));
 
     LOG.info("Update Table TTL");
     session.execute(String.format("ALTER TABLE %s WITH default_time_to_live=2;", tableName));
@@ -387,7 +387,7 @@ public void testRowTTLAfterAlter() throws Exception {
     session.execute(String.format("ALTER TABLE %s WITH default_time_to_live=20;", tableName));
 
     LOG.info("Wait for rows to expire.");
-    TestUtils.waitForTTL(8000L);
+    TestUtils.waitForTTL(4000L);
 
     LOG.info("Verify second row is gone.");
     assertNoRow(tableName, 4);

src/yb/consensus/consensus.h

@@ -147,6 +147,8 @@ struct LeaderElectionData {
 
   TEST_SuppressVoteRequest suppress_vote_request = TEST_SuppressVoteRequest::kFalse;
 
+  bool initial_election = false;
+
   std::string ToString() const;
 };
 

src/yb/consensus/consensus.proto

@@ -476,6 +476,8 @@ message RunLeaderElectionRequestPB {
   optional bytes originator_uuid = 4;
 
   optional bool suppress_vote_request = 5;
+
+  optional bool initial_election = 6;
 }
 
 message RunLeaderElectionResponsePB {

src/yb/consensus/raft_consensus.cc

@@ -208,8 +208,8 @@ DEFINE_test_flag(int32, log_change_config_every_n, 1,
 
 DEFINE_bool(enable_lease_revocation, true, "Enables lease revocation mechanism");
 
-DEFINE_bool(quick_leader_election_on_create, true, "Do we trigger quick leader elections on table "
-                                                   "creation.");
+DEFINE_bool(quick_leader_election_on_create, false,
+            "Do we trigger quick leader elections on table creation.");
 TAG_FLAG(quick_leader_election_on_create, advanced);
 TAG_FLAG(quick_leader_election_on_create, hidden);
 
@@ -511,6 +511,11 @@ Status RaftConsensus::DoStartElection(const LeaderElectionData& data, PreElected
     ReplicaState::UniqueLock lock;
     RETURN_NOT_OK(state_->LockForConfigChange(&lock));
 
+    if (data.initial_election && state_->GetCurrentTermUnlocked() != 0) {
+      LOG_WITH_PREFIX(INFO) << "Not starting initial " << election_name << " -- non zero term";
+      return Status::OK();
+    }
+
     RaftPeerPB::Role active_role = state_->GetActiveRoleUnlocked();
     if (active_role == RaftPeerPB::LEADER) {
       LOG_WITH_PREFIX(INFO) << "Not starting " << election_name << " -- already leader";

src/yb/integration-tests/cluster_itest_util.cc

@@ -196,9 +196,13 @@ vector<TServerDetails*> TServerDetailsVector(const TabletServerMapUnowned& table
   return result;
 }
 
-TabletServerMapUnowned CreateTabletServerMapUnowned(const TabletServerMap& tablet_servers) {
+TabletServerMapUnowned CreateTabletServerMapUnowned(const TabletServerMap& tablet_servers,
+                                                    const std::set<std::string>& exclude) {
   TabletServerMapUnowned result;
   for (auto& pair : tablet_servers) {
+    if (exclude.find(pair.first) != exclude.end()) {
+      continue;
+    }
     result.emplace(pair.first, pair.second.get());
   }
   return result;

src/yb/integration-tests/cluster_itest_util.h

@@ -140,7 +140,8 @@ vector<TServerDetails*> TServerDetailsVector(const TabletServerMap& tablet_serve
 vector<TServerDetails*> TServerDetailsVector(const TabletServerMapUnowned& tablet_servers);
 
 // Creates copy of tablet server map, which does not own TServerDetails.
-TabletServerMapUnowned CreateTabletServerMapUnowned(const TabletServerMap& tablet_servers);
+TabletServerMapUnowned CreateTabletServerMapUnowned(const TabletServerMap& tablet_servers,
+                                                    const std::set<std::string>& exclude = {});
 
 // Wait until all of the servers have converged on the same log index.
 // The converged index must be at least equal to 'minimum_index'.

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

@@ -708,7 +708,8 @@ TEST_F(DeleteTableTest, TestMergeConsensusMetadata) {
   };
 
   std::vector<std::string> master_flags = {
-    "--catalog_manager_wait_for_new_tablets_to_elect_leader=false"
+    "--catalog_manager_wait_for_new_tablets_to_elect_leader=false"s,
+    "--use_create_table_leader_hint=false"s,
   };
   ASSERT_NO_FATALS(StartCluster(ts_flags, master_flags));
   const MonoDelta timeout = MonoDelta::FromSeconds(10);
@@ -826,10 +827,14 @@ TEST_F(DeleteTableTest, TestDeleteFollowerWithReplicatingOperation) {
   const MonoDelta timeout = MonoDelta::FromSeconds(10);
 
   const int kNumTabletServers = 5;
-  vector<string> ts_flags, master_flags;
-  ts_flags.push_back("--enable_leader_failure_detection=false");
-  ts_flags.push_back("--maintenance_manager_polling_interval_ms=100");
-  master_flags.push_back("--catalog_manager_wait_for_new_tablets_to_elect_leader=false");
+  std::vector<std::string> ts_flags = {
+    "--enable_leader_failure_detection=false"s,
+    "--maintenance_manager_polling_interval_ms=100"s,
+  };
+  std::vector<std::string> master_flags = {
+    "--catalog_manager_wait_for_new_tablets_to_elect_leader=false"s,
+    "--use_create_table_leader_hint=false"s,
+  };
   ASSERT_NO_FATALS(StartCluster(ts_flags, master_flags, kNumTabletServers));
 
   const int kTsIndex = 0;  // We'll test with the first TS.
@@ -937,11 +942,14 @@ TEST_F(DeleteTableTest, TestMemtableNoFlushOnTabletDelete) {
 // Test that orphaned blocks are cleared from the superblock when a tablet is tombstoned.
 TEST_F(DeleteTableTest, TestOrphanedBlocksClearedOnDelete) {
   const MonoDelta timeout = MonoDelta::FromSeconds(30);
-  vector<string> ts_flags, master_flags;
-  ts_flags.push_back("--enable_leader_failure_detection=false");
-  // Flush quickly since we wait for a flush to occur.
-  ts_flags.push_back("--maintenance_manager_polling_interval_ms=100");
-  master_flags.push_back("--catalog_manager_wait_for_new_tablets_to_elect_leader=false");
+  std::vector<std::string> ts_flags = {
+    "--enable_leader_failure_detection=false"s,
+    "--maintenance_manager_polling_interval_ms=100"s,
+  };
+  std::vector<std::string> master_flags = {
+    "--catalog_manager_wait_for_new_tablets_to_elect_leader=false"s,
+    "--use_create_table_leader_hint=false"s,
+  };
   ASSERT_NO_FATALS(StartCluster(ts_flags, master_flags));
 
   const int kFollowerIndex = 0;