CASSJAVA-92: Local DC provided for nodetool clientstats

core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java

@@ -1035,7 +1035,13 @@ public enum DefaultDriverOption implements DriverOption {
    *
    * <p>Value-Type: boolean
    */
-  ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses");
+  ADDRESS_TRANSLATOR_RESOLVE_ADDRESSES("advanced.address-translator.resolve-addresses"),
+  /**
+   * Option to include driver configuration parameters in startup message.
+   *
+   * <p>Value-type: boolean
+   */
+  REQUEST_STARTUP_BAGGAGE_ENABLED("advanced.request.startup-baggage-enabled");
 
   private final String path;
 

core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java

@@ -383,6 +383,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) {
     map.put(TypedDriverOption.METRICS_GENERATE_AGGREGABLE_HISTOGRAMS, true);
     map.put(
         TypedDriverOption.LOAD_BALANCING_DC_FAILOVER_PREFERRED_REMOTE_DCS, ImmutableList.of(""));
+    map.put(TypedDriverOption.REQUEST_STARTUP_BAGGAGE_ENABLED, true);
   }
 
   @Immutable

core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java

@@ -920,6 +920,11 @@ public String toString() {
               DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_PREFERRED_REMOTE_DCS,
               GenericType.listOf(String.class));
 
+  /** Option to include driver configuration parameters in startup message. */
+  public static final TypedDriverOption<Boolean> REQUEST_STARTUP_BAGGAGE_ENABLED =
+      new TypedDriverOption<>(
+          DefaultDriverOption.REQUEST_STARTUP_BAGGAGE_ENABLED, GenericType.BOOLEAN);
+
   private static Iterable<TypedDriverOption<?>> introspectBuiltInValues() {
     try {
       ImmutableList.Builder<TypedDriverOption<?>> result = ImmutableList.builder();

core/src/main/java/com/datastax/oss/driver/api/core/loadbalancing/LoadBalancingPolicy.java

@@ -24,6 +24,7 @@
 import com.datastax.oss.driver.api.core.tracker.RequestTracker;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
+import java.util.Collections;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Queue;
@@ -76,6 +77,12 @@ default Optional<RequestTracker> getRequestTracker() {
    */
   void init(@NonNull Map<UUID, Node> nodes, @NonNull DistanceReporter distanceReporter);
 
+  /** Returns map containing details that impact C* node connectivity. */
+  @NonNull
+  default Map<String, ?> getStartupConfiguration() {
+    return Collections.emptyMap();
+  }
+
   /**
    * Returns the coordinators to use for a new query.
    *

core/src/main/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContext.java

@@ -216,8 +216,8 @@ public class DefaultDriverContext implements InternalDriverContext {
       new LazyReference<>("metricIdGenerator", this::buildMetricIdGenerator, cycleDetector);
   private final LazyReference<RequestThrottler> requestThrottlerRef =
       new LazyReference<>("requestThrottler", this::buildRequestThrottler, cycleDetector);
-  private final LazyReference<Map<String, String>> startupOptionsRef =
-      new LazyReference<>("startupOptions", this::buildStartupOptions, cycleDetector);
+  private final LazyReference<StartupOptionsBuilder> startupOptionsRef =
+      new LazyReference<>("startupOptionsFactory", this::buildStartupOptionsFactory, cycleDetector);
   private final LazyReference<NodeStateListener> nodeStateListenerRef;
   private final LazyReference<SchemaChangeListener> schemaChangeListenerRef;
   private final LazyReference<RequestTracker> requestTrackerRef;
@@ -335,16 +335,15 @@ public DefaultDriverContext(
   }
 
   /**
-   * Builds a map of options to send in a Startup message.
+   * Returns builder of options to send in a Startup message.
    *
    * @see #getStartupOptions()
    */
-  protected Map<String, String> buildStartupOptions() {
+  protected StartupOptionsBuilder buildStartupOptionsFactory() {
     return new StartupOptionsBuilder(this)
         .withClientId(startupClientId)
         .withApplicationName(startupApplicationName)
-        .withApplicationVersion(startupApplicationVersion)
-        .build();
+        .withApplicationVersion(startupApplicationVersion);
   }
 
   protected Map<String, LoadBalancingPolicy> buildLoadBalancingPolicies() {
@@ -1013,7 +1012,8 @@ public ProtocolVersion getProtocolVersion() {
   @NonNull
   @Override
   public Map<String, String> getStartupOptions() {
-    return startupOptionsRef.get();
+    // startup options are calculated dynamically and maj vary per connection
+    return startupOptionsRef.get().build();
   }
 
   protected RequestLogFormatter buildRequestLogFormatter() {

core/src/main/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilder.java

@@ -18,25 +18,36 @@
 package com.datastax.oss.driver.internal.core.context;
 
 import com.datastax.dse.driver.api.core.config.DseDriverOption;
+import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
+import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy;
 import com.datastax.oss.driver.api.core.session.Session;
 import com.datastax.oss.driver.api.core.uuid.Uuids;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import com.datastax.oss.protocol.internal.request.Startup;
 import com.datastax.oss.protocol.internal.util.collection.NullAllowingImmutableMap;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import edu.umd.cs.findbugs.annotations.Nullable;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 import net.jcip.annotations.Immutable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Immutable
 public class StartupOptionsBuilder {
 
   public static final String DRIVER_NAME_KEY = "DRIVER_NAME";
   public static final String DRIVER_VERSION_KEY = "DRIVER_VERSION";
+  public static final String DRIVER_BAGGAGE = "DRIVER_BAGGAGE";
   public static final String APPLICATION_NAME_KEY = "APPLICATION_NAME";
   public static final String APPLICATION_VERSION_KEY = "APPLICATION_VERSION";
   public static final String CLIENT_ID_KEY = "CLIENT_ID";
 
+  private static final Logger LOG = LoggerFactory.getLogger(StartupOptionsBuilder.class);
+  private static final ObjectMapper mapper = new ObjectMapper();
+
   protected final InternalDriverContext context;
   private UUID clientId;
   private String applicationName;
@@ -119,6 +130,12 @@ public Map<String, String> build() {
     if (applicationVersion != null) {
       builder.put(APPLICATION_VERSION_KEY, applicationVersion);
     }
+    boolean baggageEnabled =
+        config.getBoolean(DefaultDriverOption.REQUEST_STARTUP_BAGGAGE_ENABLED, true);
+    if (baggageEnabled) {
+      // do not cache local DC as it can change within LBP implementation
+      driverBaggage().ifPresent(s -> builder.put(DRIVER_BAGGAGE, s));
+    }
 
     return builder.build();
   }
@@ -142,4 +159,21 @@ protected String getDriverName() {
   protected String getDriverVersion() {
     return Session.OSS_DRIVER_COORDINATES.getVersion().toString();
   }
+
+  private Optional<String> driverBaggage() {
+    ImmutableMap.Builder<String, Object> builder = new ImmutableMap.Builder<>();
+    for (Map.Entry<String, LoadBalancingPolicy> entry :
+        context.getLoadBalancingPolicies().entrySet()) {
+      Map<String, ?> config = entry.getValue().getStartupConfiguration();
+      if (!config.isEmpty()) {
+        builder.put(entry.getKey(), config);
+      }
+    }
+    try {
+      return Optional.of(mapper.writeValueAsString(builder.build()));
+    } catch (Exception e) {
+      LOG.warn("Failed to construct startup driver baggage", e);
+      return Optional.empty();
+    }
+  }
 }

core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/BasicLoadBalancingPolicy.java

@@ -45,6 +45,7 @@
 import com.datastax.oss.driver.internal.core.util.collection.QueryPlan;
 import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan;
 import com.datastax.oss.driver.shaded.guava.common.base.Predicates;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import com.datastax.oss.driver.shaded.guava.common.collect.Lists;
 import com.datastax.oss.driver.shaded.guava.common.collect.Sets;
 import edu.umd.cs.findbugs.annotations.NonNull;
@@ -155,10 +156,38 @@ public BasicLoadBalancingPolicy(@NonNull DriverContext context, @NonNull String
    * Before initialization, this method always returns null.
    */
   @Nullable
-  protected String getLocalDatacenter() {
+  public String getLocalDatacenter() {
     return localDc;
   }
 
+  @NonNull
+  @Override
+  public Map<String, ?> getStartupConfiguration() {
+    ImmutableMap.Builder<String, Object> builder = new ImmutableMap.Builder<>();
+    if (localDc != null) {
+      builder.put("localDc", localDc);
+    } else {
+      // Local data center may not be discovered prior to connection pool initialization.
+      // In such scenario, return configured local data center name.
+      // Note that when using DC inferring load balancing policy, startup configuration
+      // may not show local DC name, because it will be discovered only once control connection
+      // is established and datacenter of contact points known.
+      Optional<String> configuredDc =
+          new OptionalLocalDcHelper(context, profile, logPrefix).configuredLocalDc();
+      configuredDc.ifPresent(d -> builder.put("localDc", d));
+    }
+    if (!preferredRemoteDcs.isEmpty()) {
+      builder.put("preferredRemoteDcs", preferredRemoteDcs);
+    }
+    if (allowDcFailoverForLocalCl) {
+      builder.put("allowDcFailoverForLocalCl", allowDcFailoverForLocalCl);
+    }
+    if (maxNodesPerRemoteDc > 0) {
+      builder.put("maxNodesPerRemoteDc", maxNodesPerRemoteDc);
+    }
+    return ImmutableMap.of(BasicLoadBalancingPolicy.class.getSimpleName(), builder.build());
+  }
+
   /** @return The nodes currently considered as live. */
   protected NodeSet getLiveNodes() {
     return liveNodes;

core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/DefaultLoadBalancingPolicy.java

@@ -34,6 +34,7 @@
 import com.datastax.oss.driver.internal.core.util.collection.QueryPlan;
 import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan;
 import com.datastax.oss.driver.shaded.guava.common.annotations.VisibleForTesting;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import com.datastax.oss.driver.shaded.guava.common.collect.MapMaker;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import edu.umd.cs.findbugs.annotations.Nullable;
@@ -350,4 +351,13 @@ private boolean hasSufficientResponses(long now) {
       return this.oldest - threshold >= 0;
     }
   }
+
+  @NonNull
+  @Override
+  public Map<String, ?> getStartupConfiguration() {
+    Map<String, ?> parent = super.getStartupConfiguration();
+    return ImmutableMap.of(
+        DefaultLoadBalancingPolicy.class.getSimpleName(),
+        parent.get(BasicLoadBalancingPolicy.class.getSimpleName()));
+  }
 }

core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java

@@ -65,20 +65,14 @@ public OptionalLocalDcHelper(
   @Override
   @NonNull
   public Optional<String> discoverLocalDc(@NonNull Map<UUID, Node> nodes) {
-    String localDc = context.getLocalDatacenter(profile.getName());
-    if (localDc != null) {
-      LOG.debug("[{}] Local DC set programmatically: {}", logPrefix, localDc);
-      checkLocalDatacenterCompatibility(localDc, context.getMetadataManager().getContactPoints());
-      return Optional.of(localDc);
-    } else if (profile.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) {
-      localDc = profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER);
-      LOG.debug("[{}] Local DC set from configuration: {}", logPrefix, localDc);
-      checkLocalDatacenterCompatibility(localDc, context.getMetadataManager().getContactPoints());
-      return Optional.of(localDc);
+    Optional<String> localDc = configuredLocalDc();
+    if (localDc.isPresent()) {
+      checkLocalDatacenterCompatibility(
+          localDc.get(), context.getMetadataManager().getContactPoints());
     } else {
       LOG.debug("[{}] Local DC not set, DC awareness will be disabled", logPrefix);
-      return Optional.empty();
     }
+    return localDc;
   }
 
   /**
@@ -138,4 +132,19 @@ protected String formatDcs(Iterable<? extends Node> nodes) {
     }
     return String.join(", ", new TreeSet<>(l));
   }
+
+  /** @return Local data center set programmatically or from configuration file. */
+  @NonNull
+  public Optional<String> configuredLocalDc() {
+    String localDc = context.getLocalDatacenter(profile.getName());
+    if (localDc != null) {
+      LOG.debug("[{}] Local DC set programmatically: {}", logPrefix, localDc);
+      return Optional.of(localDc);
+    } else if (profile.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) {
+      localDc = profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER);
+      LOG.debug("[{}] Local DC set from configuration: {}", logPrefix, localDc);
+      return Optional.of(localDc);
+    }
+    return Optional.empty();
+  }
 }

core/src/main/java/com/datastax/oss/driver/internal/core/util/Strings.java

@@ -81,6 +81,18 @@ public static String doubleQuote(String value) {
     return quote(value, '"');
   }
 
+  /**
+   * Double quote the given string; double quotes are escaped. If the given string is null, this
+   * method returns ({@code null}).
+   *
+   * @param value The value to double quote.
+   * @return The double quoted string.
+   */
+  public static String doubleQuoteNullable(String value) {
+    if (value == null) return null;
+    return quote(value, '"');
+  }
+
   /**
    * Unquote the given string if it is double quoted; double quotes are unescaped. If the given
    * string is not double quoted, it is returned without any modification.

core/src/main/resources/reference.conf

@@ -1150,6 +1150,14 @@ datastax-java-driver {
     # Overridable in a profile: no
     warn-if-set-keyspace = true
 
+    # Enable publishing driver baggage with Startup message that contains
+    # driver configuration details.
+    #
+    # Required: no
+    # Modifiable at runtime: yes
+    # Overridable in a profile: no
+    startup-baggage-enabled = true
+
     # If tracing is enabled for a query, this controls how the trace is fetched.
     trace {
       # How many times the driver will attempt to fetch the query if it is not ready yet.

core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/FixedHostNameAddressTranslatorTest.java

@@ -26,7 +26,6 @@
 import com.datastax.oss.driver.internal.core.context.DefaultDriverContext;
 import com.datastax.oss.driver.internal.core.context.MockedDriverContextFactory;
 import java.net.InetSocketAddress;
-import java.util.Optional;
 import org.junit.Test;
 
 public class FixedHostNameAddressTranslatorTest {
@@ -36,7 +35,7 @@ public void should_translate_address() {
     DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
     when(defaultProfile.getString(ADDRESS_TRANSLATOR_ADVERTISED_HOSTNAME)).thenReturn("myaddress");
     DefaultDriverContext defaultDriverContext =
-        MockedDriverContextFactory.defaultDriverContext(Optional.of(defaultProfile));
+        MockedDriverContextFactory.defaultDriverContext(defaultProfile);
 
     FixedHostNameAddressTranslator translator =
         new FixedHostNameAddressTranslator(defaultDriverContext);

core/src/test/java/com/datastax/oss/driver/internal/core/addresstranslation/SubnetAddressTranslatorTest.java

@@ -30,7 +30,6 @@
 import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import java.net.InetSocketAddress;
 import java.util.Map;
-import java.util.Optional;
 import org.junit.Test;
 
 @SuppressWarnings("resource")
@@ -148,6 +147,6 @@ public void should_fail_on_default_address_without_port() {
   private static DefaultDriverContext context(Map<String, String> subnetAddresses) {
     DriverExecutionProfile profile = mock(DriverExecutionProfile.class);
     when(profile.getStringMap(ADDRESS_TRANSLATOR_SUBNET_ADDRESSES)).thenReturn(subnetAddresses);
-    return MockedDriverContextFactory.defaultDriverContext(Optional.of(profile));
+    return MockedDriverContextFactory.defaultDriverContext(profile);
   }
 }

core/src/test/java/com/datastax/oss/driver/internal/core/context/DefaultDriverContextTest.java

@@ -42,7 +42,7 @@ private DefaultDriverContext buildMockedContext(Optional<String> compressionOpti
     DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
     when(defaultProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
         .thenReturn(compressionOption.orElse("none"));
-    return MockedDriverContextFactory.defaultDriverContext(Optional.of(defaultProfile));
+    return MockedDriverContextFactory.defaultDriverContext(defaultProfile);
   }
 
   private void doCreateCompressorTest(Optional<String> configVal, Class<?> expectedClz) {

core/src/test/java/com/datastax/oss/driver/internal/core/context/MockedDriverContextFactory.java

@@ -24,44 +24,45 @@
 import com.datastax.oss.driver.api.core.config.DriverConfig;
 import com.datastax.oss.driver.api.core.config.DriverConfigLoader;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
+import com.datastax.oss.driver.api.core.loadbalancing.LoadBalancingPolicy;
+import com.datastax.oss.driver.api.core.loadbalancing.NodeDistanceEvaluator;
+import com.datastax.oss.driver.api.core.metadata.Node;
 import com.datastax.oss.driver.api.core.metadata.NodeStateListener;
 import com.datastax.oss.driver.api.core.metadata.schema.SchemaChangeListener;
 import com.datastax.oss.driver.api.core.session.ProgrammaticArguments;
 import com.datastax.oss.driver.api.core.tracker.RequestTracker;
+import com.datastax.oss.driver.internal.core.ConsistencyLevelRegistry;
+import com.datastax.oss.driver.internal.core.loadbalancing.DefaultLoadBalancingPolicy;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap;
 import com.datastax.oss.driver.shaded.guava.common.collect.Maps;
+import edu.umd.cs.findbugs.annotations.NonNull;
+import edu.umd.cs.findbugs.annotations.Nullable;
 import java.time.Duration;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
 import java.util.Optional;
+import java.util.UUID;
 
 public class MockedDriverContextFactory {
 
   public static DefaultDriverContext defaultDriverContext() {
-    return defaultDriverContext(Optional.empty());
+    return defaultDriverContext(MockedDriverContextFactory.defaultProfile("datacenter1"));
   }
 
   public static DefaultDriverContext defaultDriverContext(
-      Optional<DriverExecutionProfile> profileOption) {
-
-    /* If the caller provided a profile use that, otherwise make a new one */
-    final DriverExecutionProfile profile =
-        profileOption.orElseGet(
-            () -> {
-              DriverExecutionProfile blankProfile = mock(DriverExecutionProfile.class);
-              when(blankProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
-                  .thenReturn("none");
-              when(blankProfile.getDuration(DefaultDriverOption.METRICS_NODE_EXPIRE_AFTER))
-                  .thenReturn(Duration.ofMinutes(5));
-              when(blankProfile.isDefined(DefaultDriverOption.METRICS_FACTORY_CLASS))
-                  .thenReturn(true);
-              when(blankProfile.getString(DefaultDriverOption.METRICS_FACTORY_CLASS))
-                  .thenReturn("DefaultMetricsFactory");
-              return blankProfile;
-            });
+      DriverExecutionProfile defaultProfile, DriverExecutionProfile... profiles) {
 
     /* Setup machinery to connect the input DriverExecutionProfile to the config loader */
     final DriverConfig driverConfig = mock(DriverConfig.class);
     final DriverConfigLoader configLoader = mock(DriverConfigLoader.class);
     when(configLoader.getInitialConfig()).thenReturn(driverConfig);
-    when(driverConfig.getDefaultProfile()).thenReturn(profile);
+    when(driverConfig.getDefaultProfile()).thenReturn(defaultProfile);
+    when(driverConfig.getProfile(defaultProfile.getName())).thenReturn(defaultProfile);
+
+    for (DriverExecutionProfile profile : profiles) {
+      when(driverConfig.getProfile(profile.getName())).thenReturn(profile);
+    }
 
     ProgrammaticArguments args =
         ProgrammaticArguments.builder()
@@ -71,6 +72,91 @@ public static DefaultDriverContext defaultDriverContext(
             .withLocalDatacenters(Maps.newHashMap())
             .withNodeDistanceEvaluators(Maps.newHashMap())
             .build();
-    return new DefaultDriverContext(configLoader, args);
+
+    return new DefaultDriverContext(configLoader, args) {
+      @NonNull
+      @Override
+      public Map<String, LoadBalancingPolicy> getLoadBalancingPolicies() {
+        ImmutableMap.Builder<String, LoadBalancingPolicy> map = ImmutableMap.builder();
+        map.put(
+            defaultProfile.getName(),
+            mockLoadBalancingPolicy(
+                this,
+                defaultProfile.getName(),
+                defaultProfile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)));
+        for (DriverExecutionProfile profile : profiles) {
+          map.put(
+              profile.getName(),
+              mockLoadBalancingPolicy(
+                  this,
+                  profile.getName(),
+                  profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)));
+        }
+        return map.build();
+      }
+
+      @NonNull
+      @Override
+      public ConsistencyLevelRegistry getConsistencyLevelRegistry() {
+        return mock(ConsistencyLevelRegistry.class);
+      }
+    };
+  }
+
+  public static DriverExecutionProfile defaultProfile(String localDc) {
+    return createProfile(DriverExecutionProfile.DEFAULT_NAME, localDc);
+  }
+
+  public static DriverExecutionProfile createProfile(String name, String localDc) {
+    DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
+    when(defaultProfile.getName()).thenReturn(name);
+    when(defaultProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
+        .thenReturn("none");
+    when(defaultProfile.getDuration(DefaultDriverOption.METRICS_NODE_EXPIRE_AFTER))
+        .thenReturn(Duration.ofMinutes(5));
+    when(defaultProfile.isDefined(DefaultDriverOption.METRICS_FACTORY_CLASS)).thenReturn(true);
+    when(defaultProfile.getString(DefaultDriverOption.METRICS_FACTORY_CLASS))
+        .thenReturn("DefaultMetricsFactory");
+    when(defaultProfile.getBoolean(DefaultDriverOption.REQUEST_STARTUP_BAGGAGE_ENABLED, true))
+        .thenReturn(true);
+    when(defaultProfile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER))
+        .thenReturn(localDc);
+    return defaultProfile;
+  }
+
+  public static void allowRemoteDcConnectivity(
+      DriverExecutionProfile profile,
+      int maxNodesPerRemoteDc,
+      boolean allowRemoteSatisfyLocalDc,
+      List<String> preferredRemoteDcs) {
+    when(profile.getInt(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_MAX_NODES_PER_REMOTE_DC))
+        .thenReturn(maxNodesPerRemoteDc);
+    when(profile.getBoolean(
+            DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_ALLOW_FOR_LOCAL_CONSISTENCY_LEVELS))
+        .thenReturn(allowRemoteSatisfyLocalDc);
+    when(profile.getStringList(DefaultDriverOption.LOAD_BALANCING_DC_FAILOVER_PREFERRED_REMOTE_DCS))
+        .thenReturn(preferredRemoteDcs);
+  }
+
+  private static LoadBalancingPolicy mockLoadBalancingPolicy(
+      DefaultDriverContext driverContext, String profile, String localDc) {
+    LoadBalancingPolicy loadBalancingPolicy =
+        new DefaultLoadBalancingPolicy(driverContext, profile) {
+          @NonNull
+          @Override
+          protected Optional<String> discoverLocalDc(@NonNull Map<UUID, Node> nodes) {
+            return Optional.of(localDc);
+          }
+
+          @NonNull
+          @Override
+          protected NodeDistanceEvaluator createNodeDistanceEvaluator(
+              @Nullable String localDc, @NonNull Map<UUID, Node> nodes) {
+            return mock(NodeDistanceEvaluator.class);
+          }
+        };
+    loadBalancingPolicy.init(
+        Collections.emptyMap(), mock(LoadBalancingPolicy.DistanceReporter.class));
+    return loadBalancingPolicy;
   }
 }

core/src/test/java/com/datastax/oss/driver/internal/core/context/StartupOptionsBuilderTest.java

@@ -26,10 +26,10 @@
 import com.datastax.oss.driver.api.core.config.DefaultDriverOption;
 import com.datastax.oss.driver.api.core.config.DriverExecutionProfile;
 import com.datastax.oss.driver.api.core.session.Session;
+import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList;
 import com.datastax.oss.protocol.internal.request.Startup;
 import com.tngtech.java.junit.dataprovider.DataProvider;
 import com.tngtech.java.junit.dataprovider.DataProviderRunner;
-import java.util.Optional;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
@@ -41,7 +41,7 @@ private DefaultDriverContext buildMockedContext(String compression) {
     DriverExecutionProfile defaultProfile = mock(DriverExecutionProfile.class);
     when(defaultProfile.getString(DefaultDriverOption.PROTOCOL_COMPRESSION, "none"))
         .thenReturn(compression);
-    return MockedDriverContextFactory.defaultDriverContext(Optional.of(defaultProfile));
+    return MockedDriverContextFactory.defaultDriverContext(defaultProfile);
   }
 
   private void assertDefaultStartupOptions(Startup startup) {
@@ -94,4 +94,44 @@ public void should_fail_to_build_startup_options_with_invalid_compression() {
               new Startup(ctx.getStartupOptions());
             });
   }
+
+  @Test
+  public void should_include_all_local_dcs_in_startup_message() {
+
+    DefaultDriverContext ctx =
+        MockedDriverContextFactory.defaultDriverContext(
+            MockedDriverContextFactory.defaultProfile("us-west-2"),
+            MockedDriverContextFactory.createProfile("oltp", "us-east-2"),
+            MockedDriverContextFactory.createProfile("olap", "eu-central-1"));
+    Startup startup = new Startup(ctx.getStartupOptions());
+    assertThat(startup.options)
+        .containsEntry(
+            StartupOptionsBuilder.DRIVER_BAGGAGE,
+            "{\"default\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"us-west-2\"}},"
+                + "\"oltp\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"us-east-2\"}},"
+                + "\"olap\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"eu-central-1\"}}}");
+  }
+
+  @Test
+  public void should_include_all_lbp_details_in_startup_message() {
+
+    DriverExecutionProfile defaultProfile = MockedDriverContextFactory.defaultProfile("dc1");
+    DriverExecutionProfile oltpProfile = MockedDriverContextFactory.createProfile("oltp", "dc1");
+    MockedDriverContextFactory.allowRemoteDcConnectivity(
+        oltpProfile, 2, true, ImmutableList.of("dc2", "dc3"));
+    DefaultDriverContext ctx =
+        MockedDriverContextFactory.defaultDriverContext(defaultProfile, oltpProfile);
+
+    Startup startup = new Startup(ctx.getStartupOptions());
+
+    assertThat(startup.options)
+        .containsEntry(
+            StartupOptionsBuilder.DRIVER_BAGGAGE,
+            "{\"default\":{\"DefaultLoadBalancingPolicy\":{\"localDc\":\"dc1\"}},"
+                + "\"oltp\":{\"DefaultLoadBalancingPolicy\":{"
+                + "\"localDc\":\"dc1\","
+                + "\"preferredRemoteDcs\":[\"dc2\",\"dc3\"],"
+                + "\"allowDcFailoverForLocalCl\":true,"
+                + "\"maxNodesPerRemoteDc\":2}}}");
+  }
 }