[#5561] Attach CDS to client initializer thread

Summary:
Async client initializer uses separate thread to perform it's job.
This thread does not have CDS attached, so it crashes in case of call to RPC thread pool.
Such call is possible when master resolution failed and such result already stored in DNS cache.

Test Plan:
ybd --gtest_filter ClientTest.BadMasterAddress
Jenkins: rebase: 2.1

Reviewers: timur

Reviewed By: timur

Subscribers: ybase

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

src/yb/client/async_initializer.cc

@@ -77,6 +77,8 @@ YBClient* AsyncClientInitialiser::client() const {
 }
 
 void AsyncClientInitialiser::InitClient() {
+  CDSAttacher attacher;
+
   LOG(INFO) << "Starting to init ybclient";
   while (!stopping_) {
     auto result = client_builder_.Build(messenger_);

src/yb/client/client-test.cc

@@ -41,6 +41,7 @@
 #include <glog/stl_logging.h>
 
 #include "yb/client/callbacks.h"
+#include "yb/client/async_initializer.h"
 #include "yb/client/client.h"
 #include "yb/client/client-internal.h"
 #include "yb/client/client-test-util.h"
@@ -86,6 +87,7 @@
 
 #include "yb/util/capabilities.h"
 #include "yb/util/metrics.h"
+#include "yb/util/net/dns_resolver.h"
 #include "yb/util/net/sockaddr.h"
 #include "yb/util/status.h"
 #include "yb/util/stopwatch.h"
@@ -2240,5 +2242,32 @@ TEST_F(ClientTest, GetNamespaceInfo) {
   ASSERT_TRUE(resp.colocated());
 }
 
+TEST_F(ClientTest, BadMasterAddress) {
+  auto messenger = ASSERT_RESULT(CreateMessenger("test-messenger"));
+  auto host = "should.not.resolve";
+
+  // Put host entry in cache.
+  ASSERT_NOK(messenger->resolver().Resolve(host));
+
+  {
+    struct TestServerOptions : public server::ServerBaseOptions {
+    };
+    TestServerOptions opts;
+    auto master_addr = std::make_shared<server::MasterAddresses>();
+    // Put several hosts, so resolve would take place.
+    master_addr->push_back({HostPort(host, 1)});
+    master_addr->push_back({HostPort(host, 2)});
+    opts.SetMasterAddresses(master_addr);
+
+    AsyncClientInitialiser async_init(
+        "test-client", /* num_reactors= */ 1, /* timeout_seconds= */ 1, "UUID", &opts,
+        /* metric_entity= */ nullptr, /* parent_mem_tracker= */ nullptr, messenger.get());
+    async_init.Start();
+    async_init.get_client_future().wait_for(1s);
+  }
+
+  messenger->Shutdown();
+}
+
 }  // namespace client
 }  // namespace yb