Fixing local_dc_ assignment and addressing review comments

The local DC is now only being applied to the DCAware policy if the
current `local_dc_` is not set. New unit tests were added to ensure that
the local DC is not overridden when set in the default or execution
profiles and also set when the local DC is not initially set.

Author: Michael Fero

cpp-driver/gtests/src/unit/tests/test_session.cpp

@@ -643,17 +643,13 @@ TEST_F(SessionUnitTest, HostListenerNodeDown) {
 }
 
 TEST_F(SessionUnitTest, LocalDcUpdatedOnPolicy) {
-  mockssandra::SimpleCluster cluster(simple(), 3, 2);
+  mockssandra::SimpleCluster cluster(simple(), 3, 1);
   ASSERT_EQ(cluster.start_all(), 0);
 
   TestHostListener::Ptr listener(new TestHostListener());
 
   Config config;
-  config.contact_points().push_back(Address("127.0.0.1", 9042));
-  config.set_use_randomized_contact_points(
-      false); // Ensure round robin order over DC for query execution
-  config.set_token_aware_routing(false);
-  config.set_load_balancing_policy(new DCAwarePolicy("dc1"));
+  config.contact_points().push_back(Address("127.0.0.4", 9042));
   config.set_cluster_metadata_resolver_factory(
       ClusterMetadataResolverFactory::Ptr(new LocalDcClusterMetadataResolverFactory("dc2")));
   config.set_host_listener(listener);
@@ -666,22 +662,168 @@ TEST_F(SessionUnitTest, LocalDcUpdatedOnPolicy) {
               listener->wait_for_event(WAIT_FOR_TIME));
     EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.4", 9042)),
               listener->wait_for_event(WAIT_FOR_TIME));
-    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::ADD_NODE, Address("127.0.0.5", 9042)),
+  }
+
+  for (int i = 0; i < 20; ++i) { // Validate the request processors are using DC2 only
+    QueryRequest::Ptr request(new QueryRequest("blah", 0));
+
+    ResponseFuture::Ptr future = session.execute(request, NULL);
+    EXPECT_TRUE(future->wait_for(WAIT_FOR_TIME));
+    EXPECT_FALSE(future->error());
+    EXPECT_EQ("127.0.0.4", future->address().to_string());
+  }
+
+  close(&session);
+
+  ASSERT_EQ(0u, listener->event_count());
+}
+
+TEST_F(SessionUnitTest, LocalDcNotOverriddenOnPolicy) {
+  mockssandra::SimpleCluster cluster(simple(), 1, 3);
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  TestHostListener::Ptr listener(new TestHostListener());
+
+  Config config;
+  config.contact_points().push_back(Address("127.0.0.1", 9042));
+  config.set_load_balancing_policy(new DCAwarePolicy("dc1"));
+  config.set_cluster_metadata_resolver_factory(
+    ClusterMetadataResolverFactory::Ptr(new LocalDcClusterMetadataResolverFactory("dc2")));
+  config.set_host_listener(listener);
+
+  Session session;
+  connect(config, &session);
+
+  { // Initial nodes available from peers table (should be DC1)
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::ADD_NODE, Address("127.0.0.1", 9042)),
               listener->wait_for_event(WAIT_FOR_TIME));
-    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.5", 9042)),
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.1", 9042)),
               listener->wait_for_event(WAIT_FOR_TIME));
   }
 
-  for (int i = 0; i < 20; ++i) { // Validate the request processors are using DC2 only
+  for (int i = 0; i < 20; ++i) { // Validate the request processors are using DC1 only
     QueryRequest::Ptr request(new QueryRequest("blah", 0));
 
     ResponseFuture::Ptr future = session.execute(request, NULL);
     EXPECT_TRUE(future->wait_for(WAIT_FOR_TIME));
     EXPECT_FALSE(future->error());
+    EXPECT_EQ("127.0.0.1", future->address().to_string());
+  }
+
+  close(&session);
+
+  ASSERT_EQ(0u, listener->event_count());
+}
+
+TEST_F(SessionUnitTest, LocalDcOverriddenOnPolicyUsingExecutionProfiles) {
+  mockssandra::SimpleCluster cluster(simple(), 3, 1);
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  TestHostListener::Ptr listener(new TestHostListener());
+
+  Config config;
+  config.contact_points().push_back(Address("127.0.0.4", 9042));
+  config.set_use_randomized_contact_points(
+    false); // Ensure round robin order over DC for query execution
+  config.set_cluster_metadata_resolver_factory(
+    ClusterMetadataResolverFactory::Ptr(new LocalDcClusterMetadataResolverFactory("dc2")));
+  config.set_host_listener(listener);
+
+  ExecutionProfile profile;
+  profile.set_load_balancing_policy(new DCAwarePolicy());
+  config.set_execution_profile("use_propagated_local_dc", &profile);
+
+  Session session;
+  connect(config, &session);
+
+  { // Initial nodes available from peers table (should be DC2)
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::ADD_NODE, Address("127.0.0.4", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.4", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+  }
+
+  for (int i = 0; i < 20; ++i) { // Validate the default profile is using DC2 only
+    QueryRequest::Ptr request(new QueryRequest("blah", 0));
 
-    String expected_address = "127.0.0.";
-    expected_address.append((i % 2 == 0) ? "4" : "5");
-    EXPECT_EQ(expected_address, future->address().to_string());
+    ResponseFuture::Ptr future = session.execute(request, NULL);
+    EXPECT_TRUE(future->wait_for(WAIT_FOR_TIME));
+    EXPECT_FALSE(future->error());
+    EXPECT_EQ("127.0.0.4", future->address().to_string());
+  }
+
+  for (int i = 0; i < 20; ++i) { // Validate the default profile is using DC2 only
+    QueryRequest::Ptr request(new QueryRequest("blah", 0));
+    request->set_execution_profile_name("use_propagated_local_dc");
+
+    ResponseFuture::Ptr future = session.execute(request, NULL);
+    EXPECT_TRUE(future->wait_for(WAIT_FOR_TIME));
+    EXPECT_FALSE(future->error());
+    EXPECT_EQ("127.0.0.4", future->address().to_string());
+  }
+
+  close(&session);
+
+  ASSERT_EQ(0u, listener->event_count());
+}
+
+TEST_F(SessionUnitTest, LocalDcNotOverriddenOnPolicyUsingExecutionProfiles) {
+  mockssandra::SimpleCluster cluster(simple(), 3, 1);
+  ASSERT_EQ(cluster.start_all(), 0);
+
+  TestHostListener::Ptr listener(new TestHostListener());
+
+  Config config;
+  config.contact_points().push_back(Address("127.0.0.4", 9042));
+  config.set_use_randomized_contact_points(
+    false); // Ensure round robin order over DC for query execution
+  config.set_cluster_metadata_resolver_factory(
+    ClusterMetadataResolverFactory::Ptr(new LocalDcClusterMetadataResolverFactory("dc2")));
+  config.set_host_listener(listener);
+
+  ExecutionProfile profile;
+  profile.set_load_balancing_policy(new DCAwarePolicy("dc1"));
+  config.set_execution_profile("use_dc1", &profile);
+
+  Session session;
+  connect(config, &session);
+
+  { // Initial nodes available from peers table (should be DC1 and DC2)
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::ADD_NODE, Address("127.0.0.1", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.1", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::ADD_NODE, Address("127.0.0.2", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.2", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::ADD_NODE, Address("127.0.0.3", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.3", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::ADD_NODE, Address("127.0.0.4", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+    EXPECT_EQ(HostEventFuture::Event(HostEventFuture::START_NODE, Address("127.0.0.4", 9042)),
+              listener->wait_for_event(WAIT_FOR_TIME));
+  }
+
+  for (int i = 0; i < 20; ++i) { // Validate the default profile is using DC2 only
+    QueryRequest::Ptr request(new QueryRequest("blah", 0));
+
+    ResponseFuture::Ptr future = session.execute(request, NULL);
+    EXPECT_TRUE(future->wait_for(WAIT_FOR_TIME));
+    EXPECT_FALSE(future->error());
+    EXPECT_EQ("127.0.0.4", future->address().to_string());
+  }
+
+  for (int i = 0; i < 20; ++i) { // Validate the default profile is using DC1 only
+    QueryRequest::Ptr request(new QueryRequest("blah", 0));
+    request->set_execution_profile_name("use_dc1");
+
+    ResponseFuture::Ptr future = session.execute(request, NULL);
+    EXPECT_TRUE(future->wait_for(WAIT_FOR_TIME));
+    EXPECT_FALSE(future->error());
+    EXPECT_NE("127.0.0.4", future->address().to_string());
   }
 
   close(&session);

cpp-driver/src/cluster_connector.cpp

@@ -116,14 +116,12 @@ void ClusterConnector::internal_resolve_and_connect() {
                      bind_callback(&ClusterConnector::on_resolve, this));
 }
 
-void ClusterConnector::internal_connect(const Address& address, ProtocolVersion version,
-                                        const String& local_dc) {
+void ClusterConnector::internal_connect(const Address& address, ProtocolVersion version) {
   Host::Ptr host(new Host(address));
   ControlConnector::Ptr connector(
       new ControlConnector(host, version, bind_callback(&ClusterConnector::on_connect, this)));
   connectors_[address] = connector; // Keep track of the connectors so they can be canceled.
   connector->with_metrics(metrics_)
-      ->with_local_dc(local_dc)
       ->with_settings(settings_.control_connection_settings)
       ->connect(event_loop_->loop());
 }
@@ -180,11 +178,12 @@ void ClusterConnector::on_resolve(ClusterMetadataResolver* resolver) {
     return;
   }
 
+  local_dc_ = resolver->local_dc();
   remaining_connector_count_ = resolved_contact_points.size();
   for (AddressVec::const_iterator it = resolved_contact_points.begin(),
                                   end = resolved_contact_points.end();
        it != end; ++it) {
-    internal_connect(*it, protocol_version_, resolver->local_dc());
+    internal_connect(*it, protocol_version_);
   }
 }
 
@@ -260,7 +259,7 @@ void ClusterConnector::on_connect(ControlConnector* connector) {
 
     cluster_.reset(new Cluster(connector->release_connection(), listener_, event_loop_,
                                connected_host, hosts, connector->schema(), default_policy, policies,
-                               connector->local_dc(), settings_));
+                               local_dc_, settings_));
 
     // Clear any connection errors and set the final negotiated protocol version.
     error_code_ = CLUSTER_OK;
@@ -282,7 +281,7 @@ void ClusterConnector::on_connect(ControlConnector* connector) {
       on_error(CLUSTER_ERROR_INVALID_PROTOCOL, "Unable to find supported protocol version");
       return;
     }
-    internal_connect(connector->address(), lower_version, connector->local_dc());
+    internal_connect(connector->address(), lower_version);
   } else if (connector->is_ssl_error()) {
     ssl_error_code_ = connector->ssl_error_code();
     on_error(CLUSTER_ERROR_SSL_ERROR, connector->error_message());

cpp-driver/src/cluster_connector.hpp

@@ -138,7 +138,7 @@ class ClusterConnector : public RefCounted<ClusterConnector> {
 
 private:
   void internal_resolve_and_connect();
-  void internal_connect(const Address& address, ProtocolVersion version, const String& local_dc);
+  void internal_connect(const Address& address, ProtocolVersion version);
   void internal_cancel();
 
   void finish();
@@ -168,6 +168,7 @@ class ClusterConnector : public RefCounted<ClusterConnector> {
   EventLoop* event_loop_;
   Random* random_;
   Metrics* metrics_;
+  String local_dc_;
   ClusterSettings settings_;
 
   Callback callback_;

cpp-driver/src/connector.cpp

@@ -201,11 +201,6 @@ Connector* Connector::with_metrics(Metrics* metrics) {
   return this;
 }
 
-Connector* Connector::with_local_dc(const String& local_dc) {
-  local_dc_ = local_dc;
-  return this;
-}
-
 Connector* Connector::with_settings(const ConnectionSettings& settings) {
   settings_ = settings;
   // Only use hostname resolution if actually required for SSL or

cpp-driver/src/connector.hpp

@@ -132,16 +132,6 @@ class Connector
    */
   Connector* with_metrics(Metrics* metrics);
 
-  /**
-   * Set the local datacenter to use for the initializing the load balancing policies after the
-   * connection is connected.
-   *
-   * @param local_dc The local datacenter determined by the metadata service for initializing the
-   * load balancing policies.
-   * @return The connector to chain calls.
-   */
-  Connector* with_local_dc(const String& local_dc);
-
   /**
    * Set the connection and socket settings.
    *
@@ -232,7 +222,6 @@ class Connector
   int event_types_;
   ConnectionListener* listener_;
   Metrics* metrics_;
-  String local_dc_;
   ConnectionSettings settings_;
 };
 

cpp-driver/src/control_connector.cpp

@@ -90,11 +90,6 @@ ControlConnector* ControlConnector::with_metrics(Metrics* metrics) {
   return this;
 }
 
-ControlConnector* ControlConnector::with_local_dc(const String& local_dc) {
-  local_dc_ = local_dc;
-  return this;
-}
-
 ControlConnector* ControlConnector::with_settings(const ControlConnectionSettings& settings) {
   settings_ = settings;
   return this;
@@ -109,7 +104,6 @@ void ControlConnector::connect(uv_loop_t* loop) {
     event_types = CASS_EVENT_TOPOLOGY_CHANGE | CASS_EVENT_STATUS_CHANGE;
   }
   connector_->with_metrics(metrics_)
-      ->with_local_dc(local_dc_)
       ->with_settings(settings_.connection_settings)
       ->with_event_types(event_types)
       ->connect(loop);

cpp-driver/src/control_connector.hpp

@@ -96,15 +96,6 @@ class ControlConnector
    */
   ControlConnector* with_metrics(Metrics* metrics);
 
-  /**
-   * Set the local datacenter to use for the connection.
-   *
-   * @param local_dc The local datacenter determined by the metadata service for initializing the
-   * load balancing policies.
-   * @return The connector to chain calls.
-   */
-  ControlConnector* with_local_dc(const String& local_dc);
-
   /**
    * Sets the control connection settings as well as the underlying settings
    * for the connection and socket.

cpp-driver/src/dc_aware_policy.cpp

@@ -43,13 +43,12 @@ DCAwarePolicy::DCAwarePolicy(const String& local_dc, size_t used_hosts_per_remot
 DCAwarePolicy::~DCAwarePolicy() { uv_rwlock_destroy(&available_rwlock_); }
 
 void DCAwarePolicy::init(const Host::Ptr& connected_host, const HostMap& hosts, Random* random,
-                         const String& local_dc /*= ""*/) {
-  if (!local_dc.empty()) {
-    if (!local_dc_.empty()) {
-      LOG_INFO("Overriding local data center %s with %s", local_dc_.c_str(), local_dc.c_str());
-    }
+                         const String& local_dc) {
+  if (local_dc_.empty()) { // Do not override
     local_dc_ = local_dc;
-  } else if (local_dc_.empty() && connected_host && !connected_host->dc().empty()) {
+  }
+
+  if (local_dc_.empty() && connected_host && !connected_host->dc().empty()) {
     LOG_INFO("Using '%s' for the local data center "
              "(if this is incorrect, please provide the correct data center)",
              connected_host->dc().c_str());

cpp-driver/src/list_policy.cpp

@@ -23,7 +23,7 @@ using namespace datastax::internal;
 using namespace datastax::internal::core;
 
 void ListPolicy::init(const Host::Ptr& connected_host, const HostMap& hosts, Random* random,
-                      const String& local_dc /*= ""*/) {
+                      const String& local_dc) {
   HostMap valid_hosts;
   for (HostMap::const_iterator i = hosts.begin(), end = hosts.end(); i != end; ++i) {
     const Host::Ptr& host = i->second;

cpp-driver/src/round_robin_policy.cpp

@@ -33,7 +33,7 @@ RoundRobinPolicy::RoundRobinPolicy()
 RoundRobinPolicy::~RoundRobinPolicy() { uv_rwlock_destroy(&available_rwlock_); }
 
 void RoundRobinPolicy::init(const Host::Ptr& connected_host, const HostMap& hosts, Random* random,
-                            const String& local_dc /*= ""*/) {
+                            const String& local_dc) {
   available_.resize(hosts.size());
   std::transform(hosts.begin(), hosts.end(), std::inserter(available_, available_.begin()),
                  GetAddress());