From ed81bf305323be6b7b0185a1c6b97c4d7f18e983 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Tue, 21 Jul 2026 00:12:30 +0200 Subject: [PATCH] fix: expand contact point hostnames to all DNS IPs at query plan time (DRIVER-201) When RESOLVE_CONTACT_POINTS=false (the default) a hostname contact point was stored as a single unresolved InetSocketAddress, so the query plan tried only the first DNS IP. Keep contact points unresolved and expand each hostname to all its DNS IPs at query-plan time via MetadataManager.getResolvedContactPoints(), so the driver falls back to the next candidate when one IP is unreachable. Resolution is bounded, concurrent and best-effort. getResolvedContactPoints() runs on the admin event loop, where nothing should block, so each blocking InetAddress.getAllByName() call is offloaded to a cached daemon-thread pool and all unresolved hostnames are resolved concurrently against a single CONTACT_POINT_RESOLUTION_TIMEOUT deadline. A cached pool (rather than one shared thread) means each hostname resolves on its own thread, so one slow or blackholed lookup cannot starve the sibling contact points, nor the next reconnect that would otherwise queue behind it. If a hostname cannot be resolved or resolution times out, the original unresolved contact point is kept as-is rather than dropped, so the query plan is never emptier than the configured contact points and the address can still be resolved later at connection time (as it was before DNS expansion existed). This is an interim mitigation, superseded by #890's non-blocking EndPoint.resolveAll(). Default advanced.control-connection.reconnection.fallback-to-original-contact-points to true (no longer Experimental): it is the DNS re-resolution path on reconnect. Metadata nodes hold an already-resolved endpoint that is never re-resolved, so falling back to the original unresolved contact points re-expands the hostname to its current DNS IPs. Document that DNS-expanded contact points are IP-backed connection candidates that may be persisted in metadata, and that each synthetic endpoint retains the original hostname (built from the resolved InetAddress) so TLS peer host / SNI / hostname verification keep using the configured hostname. Gate the control-connection reconnection contact-point fallback behind a new TopologyMonitor.reresolvesNodeAddresses() (default false; true for the proxy-based ClientRoutesTopologyMonitor and CloudTopologyMonitor). Those monitors reach nodes through endpoints that already re-resolve on every connection attempt and maintain an authoritative node set, so appending raw contact points to their reconnection plan is unnecessary and could resurrect removed nodes (PrivateLink/Cloud regression safety). The reconnection plan also appends the contact points only once the load balancing policy is RUNNING, so the pre-init plan (already built from the resolved contact points) is not duplicated or re-resolved. Remove OptionalLocalDcHelper.checkLocalDatacenterCompatibility(): it warned when a contact point reported a different datacenter than the configured local DC. Since commit 12e6acb90b switched initial metadata refresh to hostId-only matching, contact-point nodes are never reused and their datacenter stays null; comparing a configured local DC against that null made the check fire as a false positive for every contact point whenever local-datacenter was set on the default profile, rather than surface a real mismatch. The node-based "configured local DC matches no node" warning (against discovered nodes whose datacenters are populated) is retained, so the only user-visible effect is that the spurious warning is no longer emitted. Co-Authored-By: Claude Opus 4.8 --- .../core/insights/InsightsClient.java | 2 +- .../api/core/config/DefaultDriverOption.java | 13 +- .../driver/api/core/config/OptionsMap.java | 2 +- .../api/core/config/TypedDriverOption.java | 10 +- .../api/core/session/SessionBuilder.java | 22 +- .../helper/OptionalLocalDcHelper.java | 74 ++----- .../metadata/ClientRoutesTopologyMonitor.java | 17 ++ .../core/metadata/CloudTopologyMonitor.java | 8 + .../metadata/LoadBalancingPolicyWrapper.java | 33 ++- .../core/metadata/MetadataManager.java | 209 ++++++++++++++++++ .../core/metadata/TopologyMonitor.java | 20 ++ core/src/main/resources/reference.conf | 18 +- .../core/insights/InsightsClientTest.java | 2 +- .../ClientRoutesTopologyMonitorTest.java | 55 +++++ .../LoadBalancingPolicyWrapperTest.java | 76 ++++++- .../core/metadata/MetadataManagerTest.java | 170 ++++++++++++++ .../driver/core/heartbeat/HeartbeatIT.java | 4 + .../driver/core/resolver/MockResolverIT.java | 42 +++- 18 files changed, 683 insertions(+), 94 deletions(-) diff --git a/core/src/main/java/com/datastax/dse/driver/internal/core/insights/InsightsClient.java b/core/src/main/java/com/datastax/dse/driver/internal/core/insights/InsightsClient.java index 168477894ed..a592336a0dc 100644 --- a/core/src/main/java/com/datastax/dse/driver/internal/core/insights/InsightsClient.java +++ b/core/src/main/java/com/datastax/dse/driver/internal/core/insights/InsightsClient.java @@ -313,7 +313,7 @@ private InsightsStartupData createStartupData() { .withDriverVersion(getDriverVersion(startupOptions)) .withContactPoints( getResolvedContactPoints( - driverContext.getMetadataManager().getContactPoints().stream() + driverContext.getMetadataManager().getResolvedContactPoints().stream() .map(n -> n.getEndPoint().resolve()) .filter(InetSocketAddress.class::isInstance) .map(InetSocketAddress.class::cast) diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java index 56a6fe9c2be..da8d021746c 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/DefaultDriverOption.java @@ -701,8 +701,13 @@ public enum DefaultDriverOption implements DriverOption { CONTROL_CONNECTION_AGREEMENT_WARN("advanced.control-connection.schema-agreement.warn-on-failure"), /** - * Whether to forcibly add original contact points held by MetadataManager to the reconnection - * plan, in case there is no live nodes available according to LBP. Experimental. + * Whether to append the original contact points held by MetadataManager to the reconnection plan, + * after the live nodes reported by the load balancing policy. Defaults to {@code true}. + * + *

This is also the driver's DNS re-resolution path: contact points are expanded to their + * current DNS IPs at query-plan time, whereas metadata nodes hold an already-resolved endpoint + * that is never re-resolved. Keeping this enabled lets control-connection reconnects re-resolve + * the original hostnames and pick up new IPs once the live-node plan is exhausted. * *

Value-type: boolean */ @@ -837,7 +842,11 @@ public enum DefaultDriverOption implements DriverOption { * Whether to resolve the addresses passed to `basic.contact-points`. * *

Value-type: boolean + * + * @deprecated Contact points are now always kept as unresolved hostnames and expanded to all + * their DNS-mapped IPs lazily at connection time. Setting this option has no effect. */ + @Deprecated RESOLVE_CONTACT_POINTS("advanced.resolve-contact-points"), /** diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java index 60190fc1cce..e3d8e4beeb3 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/OptionsMap.java @@ -369,7 +369,7 @@ protected static void fillWithDriverDefaults(OptionsMap map) { map.put(TypedDriverOption.CONTROL_CONNECTION_AGREEMENT_INTERVAL, Duration.ofMillis(200)); map.put(TypedDriverOption.CONTROL_CONNECTION_AGREEMENT_TIMEOUT, Duration.ofSeconds(10)); map.put(TypedDriverOption.CONTROL_CONNECTION_AGREEMENT_WARN, true); - map.put(TypedDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, false); + map.put(TypedDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, true); map.put(TypedDriverOption.PREPARE_ON_ALL_NODES, true); map.put(TypedDriverOption.REPREPARE_ENABLED, true); map.put(TypedDriverOption.REPREPARE_CHECK_SYSTEM_TABLE, false); diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java index e7c606cade8..fa12f40fd3e 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java @@ -600,7 +600,7 @@ public String toString() { public static final TypedDriverOption CONTROL_CONNECTION_AGREEMENT_WARN = new TypedDriverOption<>( DefaultDriverOption.CONTROL_CONNECTION_AGREEMENT_WARN, GenericType.BOOLEAN); - /** Whether to forcibly try original contacts if no live nodes are available */ + /** Whether to append the original contact points to the reconnection plan (defaults to true) */ public static final TypedDriverOption CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS = new TypedDriverOption<>( DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, GenericType.BOOLEAN); @@ -664,7 +664,13 @@ public String toString() { /** The coalescer reschedule interval. */ public static final TypedDriverOption COALESCER_INTERVAL = new TypedDriverOption<>(DefaultDriverOption.COALESCER_INTERVAL, GenericType.DURATION); - /** Whether to resolve the addresses passed to `basic.contact-points`. */ + /** + * Whether to resolve the addresses passed to `basic.contact-points`. + * + * @deprecated Contact points are now always kept as unresolved hostnames and expanded to all + * their DNS-mapped IPs lazily at connection time. Setting this option has no effect. + */ + @Deprecated public static final TypedDriverOption RESOLVE_CONTACT_POINTS = new TypedDriverOption<>(DefaultDriverOption.RESOLVE_CONTACT_POINTS, GenericType.BOOLEAN); /** diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java b/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java index 8375f0ef30b..e4937f60d1d 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java @@ -166,11 +166,15 @@ protected DriverConfigLoader defaultConfigLoader(@Nullable ClassLoader classLoad *

Contact points can also be provided statically in the configuration. If both are specified, * they will be merged. If both are absent, the driver will default to 127.0.0.1:9042. * - *

Contrary to the configuration, DNS names with multiple A-records will not be handled here. - * If you need that, extract them manually with {@link java.net.InetAddress#getAllByName(String)} - * before calling this method. Similarly, if you need connect addresses to stay unresolved, make - * sure you pass unresolved instances here (see {@code advanced.resolve-contact-points} in the - * configuration for more explanations). + *

The driver automatically expands any contact point backed by an unresolved hostname to all + * its DNS-mapped IPs at query plan time (via {@code MetadataManager.getResolvedContactPoints()}), + * so passing a single hostname is sufficient to try all its IPs on initial connect. This applies + * equally to hostnames provided here programmatically (build an unresolved {@link + * InetSocketAddress} with {@link InetSocketAddress#createUnresolved(String, int)} to opt in) and + * to hostnames specified in the configuration. An already-resolved address passed here (the + * common case when constructing an {@code InetSocketAddress} directly from a hostname, which + * resolves eagerly) is used as provided, with no further expansion. The {@code + * advanced.resolve-contact-points} option is deprecated and has no effect. */ @NonNull public SelfT addContactPoints(@NonNull Collection contactPoints) { @@ -957,11 +961,11 @@ protected final CompletionStage buildDefaultSessionAsync() { programmaticArguments = programmaticArgumentsBuilder.build(); } - boolean resolveAddresses = - defaultConfig.getBoolean(DefaultDriverOption.RESOLVE_CONTACT_POINTS, false); - + // RESOLVE_CONTACT_POINTS is deprecated: contact points are always kept as unresolved + // hostnames and expanded to all their DNS IPs at query plan time (before the first + // connection attempt) via MetadataManager.getResolvedContactPoints(). Set contactPoints = - ContactPoints.merge(programmaticContactPoints, configContactPoints, resolveAddresses); + ContactPoints.merge(programmaticContactPoints, configContactPoints, false); if (keyspace == null && defaultConfig.isDefined(DefaultDriverOption.SESSION_KEYSPACE)) { keyspace = diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java index b93a16a6525..97aab92fff7 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/loadbalancing/helper/OptionalLocalDcHelper.java @@ -26,7 +26,6 @@ import edu.umd.cs.findbugs.annotations.NonNull; import java.util.ArrayList; import java.util.HashSet; -import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -68,73 +67,34 @@ public OptionalLocalDcHelper( @Override @NonNull public Optional discoverLocalDc(@NonNull Map nodes) { - String localDcStr = context.getLocalDatacenter(profile.getName()); - Optional localDc; - if (localDcStr != null) { - LOG.debug("[{}] Local DC set programmatically: {}", logPrefix, localDcStr); - localDc = Optional.of(localDcStr); + String localDc = context.getLocalDatacenter(profile.getName()); + if (localDc != null) { + LOG.debug("[{}] Local DC set programmatically: {}", logPrefix, localDc); } else if (profile.isDefined(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER)) { - localDcStr = profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER); - LOG.debug("[{}] Local DC set from configuration: {}", logPrefix, localDcStr); - localDc = Optional.of(localDcStr); - } else { - localDc = Optional.empty(); - } - if (localDc.isPresent()) { - checkLocalDatacenterCompatibility( - localDc.get(), context.getMetadataManager().getContactPoints()); - // Also warn if the configured DC doesn't match any node in the cluster - if (!nodes.isEmpty()) { - boolean found = false; - for (Node node : nodes.values()) { - if (localDc.get().equals(node.getDatacenter())) { - found = true; - break; - } - } - if (!found) { - LOG.warn( - "[{}] Configured local DC '{}' does not match any node's datacenter" - + " (available DCs: {}); please verify your configuration", - logPrefix, - localDc.get(), - formatDcs(nodes.values())); - } - } + localDc = profile.getString(DefaultDriverOption.LOAD_BALANCING_LOCAL_DATACENTER); + LOG.debug("[{}] Local DC set from configuration: {}", logPrefix, localDc); } else { LOG.debug("[{}] Local DC not set, DC awareness will be disabled", logPrefix); + return Optional.empty(); } - return localDc; - } - - /** - * Checks if the contact points are compatible with the local datacenter specified either through - * configuration, or programmatically. - * - *

The default implementation logs a warning when a contact point reports a datacenter - * different from the local one, and only for the default profile. - * - * @param localDc The local datacenter, as specified in the config, or programmatically. - * @param contactPoints The contact points provided when creating the session. - */ - protected void checkLocalDatacenterCompatibility( - @NonNull String localDc, Set contactPoints) { - if (profile.getName().equals(DriverExecutionProfile.DEFAULT_NAME)) { - Set badContactPoints = new LinkedHashSet<>(); - for (Node node : contactPoints) { - if (!Objects.equals(localDc, node.getDatacenter())) { - badContactPoints.add(node); + if (!nodes.isEmpty()) { + boolean found = false; + for (Node node : nodes.values()) { + if (localDc.equals(node.getDatacenter())) { + found = true; + break; } } - if (!badContactPoints.isEmpty()) { + if (!found) { LOG.warn( - "[{}] You specified {} as the local DC, but some contact points are from a different DC: {}; " - + "please provide the correct local DC, or check your contact points", + "[{}] Configured local DC '{}' does not match any node's datacenter" + + " (available DCs: {}); please verify your configuration", logPrefix, localDc, - formatNodesAndDcs(badContactPoints)); + formatDcs(nodes.values())); } } + return Optional.of(localDc); } /** diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java index 1ffc35fd9f4..a93c34463bd 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java @@ -21,6 +21,7 @@ import com.datastax.oss.driver.api.core.config.ClientRoutesConfig; import com.datastax.oss.driver.api.core.config.DefaultDriverOption; import com.datastax.oss.driver.api.core.metadata.EndPoint; +import com.datastax.oss.driver.api.core.metadata.Node; import com.datastax.oss.driver.internal.core.adminrequest.AdminRequestHandler; import com.datastax.oss.driver.internal.core.adminrequest.AdminResult; import com.datastax.oss.driver.internal.core.adminrequest.AdminRow; @@ -480,6 +481,22 @@ protected EndPoint buildNodeEndPoint( return new ClientRoutesEndPoint(this, hostId, broadcastInetAddress, fallback); } + @Override + public boolean reresolvesNodeAddresses() { + // ClientRoutesEndPoint re-resolves via the client route hostname on every connection attempt, + // but only when a route exists for that host_id (see ClientRoutesEndPoint#resolve()) -- for + // mixed/incomplete route sets it falls back to a static, non-re-resolving endpoint instead. + // Only report true when every currently-known node actually has a live route; otherwise the + // contact-point reconnection fallback must stay available for the nodes stuck on the fallback. + Map routes = resolvedRoutesCache.get(); + for (Node node : context.getMetadataManager().getMetadata().getNodes().values()) { + if (!routes.containsKey(node.getHostId())) { + return false; + } + } + return true; + } + /** * Builds the CQL query to fetch client routes. * diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java index 021824a9b16..013027e139d 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java @@ -44,4 +44,12 @@ protected EndPoint buildNodeEndPoint( UUID hostId = Objects.requireNonNull(row.getUuid("host_id")); return new SniEndPoint(cloudProxyAddress, hostId.toString()); } + + @Override + public boolean reresolvesNodeAddresses() { + // Nodes are reached through the cloud SNI proxy via SniEndPoint, which re-resolves the proxy + // hostname on every connection attempt. Appending the original contact points as a DNS + // re-resolution fallback is therefore unnecessary and could resurrect removed nodes. + return true; + } } diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapper.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapper.java index f3f3e4fe346..935a81ab883 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapper.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapper.java @@ -147,8 +147,7 @@ public Queue newQueryPlan( switch (stateRef.get()) { case BEFORE_INIT: case DURING_INIT: - // The contact points are not stored in the metadata yet: - List nodes = new ArrayList<>(context.getMetadataManager().getContactPoints()); + List nodes = new ArrayList<>(context.getMetadataManager().getResolvedContactPoints()); Collections.shuffle(nodes); return new ConcurrentLinkedQueue<>(nodes); case RUNNING: @@ -164,15 +163,33 @@ public Queue newQueryPlan( @NonNull public Queue newControlReconnectionQueryPlan() { + // Read the state once, before building the regular plan. State transitions are monotonic + // (BEFORE_INIT -> DURING_INIT -> RUNNING -> ...), so this captured value is <= the value + // newQueryPlan() reads internally; that guarantees we never both build the plan from resolved + // contact points (pre-RUNNING branch of newQueryPlan) and append them again below. + State state = stateRef.get(); Queue regularQueryPlan = newQueryPlan(null, DriverExecutionProfile.DEFAULT_NAME, null); - if (context - .getConfig() - .getDefaultProfile() - .getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS)) { - Set originalNodes = context.getMetadataManager().getContactPoints(); + // Only append resolved contact points as an explicit fallback once the LBP is RUNNING: before + // that (BEFORE_INIT/DURING_INIT), newQueryPlan() above already built regularQueryPlan directly + // from getResolvedContactPoints(), so appending them again here would just duplicate every + // entry and re-trigger DNS resolution for no benefit. + if (state == State.RUNNING + && context + .getConfig() + .getDefaultProfile() + .getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS) + && !context.getTopologyMonitor().reresolvesNodeAddresses()) { + // Use DNS-expanded contact points so every resolved IP is tried as a fallback, not just + // the first IP the JVM happens to return for the hostname. + // + // Skipped when the topology monitor re-resolves node addresses on its own (e.g. proxy-based + // monitors such as client routes or the cloud SNI proxy): those keep addresses fresh without + // this fallback, and appending raw contact points could resurrect nodes the monitor has + // authoritatively removed. + List resolvedNodes = context.getMetadataManager().getResolvedContactPoints(); List contactNodes = new ArrayList<>(); - for (DefaultNode node : originalNodes) { + for (Node node : resolvedNodes) { contactNodes.add(DefaultNode.newContactPoint(node.getEndPoint(), context)); } Collections.shuffle(contactNodes); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java index cd765c818e6..c914c9a0b5e 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/MetadataManager.java @@ -49,8 +49,12 @@ import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet; import edu.umd.cs.findbugs.annotations.NonNull; import io.netty.util.concurrent.EventExecutor; +import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.UnknownHostException; import java.nio.ByteBuffer; +import java.time.Duration; +import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -59,6 +63,14 @@ import java.util.UUID; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; import net.jcip.annotations.ThreadSafe; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -71,6 +83,11 @@ public class MetadataManager implements AsyncAutoCloseable { static final EndPoint DEFAULT_CONTACT_POINT = new DefaultEndPoint(new InetSocketAddress("127.0.0.1", 9042)); + // Bounds how long a single contact-point hostname resolution may block the caller (typically the + // admin event loop, see getResolvedContactPoints()). Not configurable on purpose: this is an + // interim mitigation, see the note on getResolvedContactPoints() below. + private static final Duration CONTACT_POINT_RESOLUTION_TIMEOUT = Duration.ofSeconds(3); + private final InternalDriverContext context; private final String logPrefix; private final EventExecutor adminExecutor; @@ -78,6 +95,16 @@ public class MetadataManager implements AsyncAutoCloseable { private final SingleThreaded singleThreaded; private final ControlConnection controlConnection; + // Contact-point hostname resolution (getResolvedContactPoints()) runs on the admin event loop, + // where nothing should block. InetAddress.getAllByName() blocks, so each resolution is offloaded + // here and bounded by CONTACT_POINT_RESOLUTION_TIMEOUT. A cached pool (rather than a single + // shared thread) resolves each hostname on its own thread: a slow or blackholed lookup that + // ignores its interrupt only ties up its own daemon thread until the OS resolver finally + // returns, instead of starving the sibling contact points -- or the next reconnect that would + // otherwise queue behind it. This is an interim mitigation, superseded by #890's non-blocking + // EndPoint.resolveAll(). + private final ExecutorService contactPointResolverExecutor; + private volatile DefaultMetadata metadata; // only updated from adminExecutor private volatile boolean schemaEnabledInConfig; private volatile List refreshedKeyspaces; @@ -107,6 +134,19 @@ protected MetadataManager(InternalDriverContext context, DefaultMetadata initial DefaultDriverOption.METADATA_SCHEMA_REFRESHED_KEYSPACES, Collections.emptyList()); this.keyspaceFilter = KeyspaceFilter.newInstance(logPrefix, refreshedKeyspaces); this.tokenMapEnabled = config.getBoolean(DefaultDriverOption.METADATA_TOKEN_MAP_ENABLED); + AtomicInteger resolverThreadCount = new AtomicInteger(); + this.contactPointResolverExecutor = + Executors.newCachedThreadPool( + runnable -> { + Thread thread = + new Thread( + runnable, + logPrefix + + "-contact-point-resolver-" + + resolverThreadCount.incrementAndGet()); + thread.setDaemon(true); + return thread; + }); context.getEventBus().register(ConfigChangeEvent.class, this::onConfigChanged); } @@ -173,6 +213,167 @@ public Set getContactPoints() { return contactPoints; } + /** + * Returns the contact points expanded to all their DNS-resolved IPs. + * + *

Contact points are stored with unresolved hostnames (the driver always uses deferred DNS + * resolution). For each contact point whose underlying address is an unresolved {@link + * InetSocketAddress}, this method calls {@link InetAddress#getAllByName(String)} to obtain every + * IP the hostname maps to and creates a synthetic contact-point {@link DefaultNode} for each IP. + * This lets the load balancing policy iterate over all candidate IPs rather than only the first + * one, so that a non-responsive IP does not block initial connection or control-connection + * reconnection. + * + *

The returned synthetic nodes are IP-backed connection candidates, not mere hostname + * wrappers: their endpoint resolves to a concrete IP. If such a node becomes the control node, + * its resolved endpoint is stored as-is in metadata (see {@link #registerNode(NodeInfo)}), so + * later reconnects keep using that IP unless they fall back to the original (unresolved) contact + * points, which re-enters this method and re-resolves DNS. + * + *

TLS note: each synthetic endpoint is built from the {@link InetAddress} returned by + * resolving the original hostname, which retains that hostname. This means the TCP connection + * uses the selected IP while the SSL engine still receives the original contact-point hostname + * for peer host / SNI / hostname verification. This must be preserved: constructing the endpoint + * from a raw IP string instead would break hostname verification and implicit SNI. + * + *

Already-resolved addresses, and endpoints that are not a {@link DefaultEndPoint} (e.g. an + * SNI or client-routes endpoint), are returned as-is: those specialized endpoint types carry + * identity beyond a plain socket address (SNI server name, host_id) and must not be naively + * reconstructed from a resolved IP. + * + *

Resolution is best-effort: if a hostname cannot be resolved, or resolution exceeds {@link + * #getContactPointResolutionTimeout()}, the original unresolved contact-point node is + * kept as-is instead of being dropped. The query plan is therefore never emptier than the + * configured contact points, and the hostname can still be resolved later at connection time (as + * it was before DNS expansion existed). + */ + public List getResolvedContactPoints() { + Set nodes = contactPoints; + if (nodes == null) { + return new ArrayList<>(); + } + List result = new ArrayList<>(); + // NOTE (interim mitigation, superseded in #890): this method is called from the + // control-connection query-plan path, i.e. the admin event loop (ControlConnection asserts + // inEventLoop()), where nothing should ever block. InetAddress.getAllByName() is a blocking + // call, so each unresolved hostname is submitted to contactPointResolverExecutor (pass 1) and + // then collected against a single shared deadline (pass 2). Resolving concurrently rather than + // one-at-a-time keeps the total wait bounded by roughly one CONTACT_POINT_RESOLUTION_TIMEOUT + // regardless of the number of contact points, and keeps one slow/blackholed hostname from + // starving the others. The calling thread still waits for the result, so this is a bound, not a + // true fix -- #890 moves multi-address resolution into the EndPoint API / ChannelFactory + // (EndPoint.resolveAll()) for proper non-blocking resolution; revisit when that lands. + List pending = new ArrayList<>(); + for (DefaultNode node : nodes) { + EndPoint endPoint = node.getEndPoint(); + if (endPoint instanceof DefaultEndPoint) { + InetSocketAddress address = ((DefaultEndPoint) endPoint).resolve(); + if (address.isUnresolved()) { + // Expand hostname to all IPs so callers can try each one in turn. + try { + Future future = + contactPointResolverExecutor.submit( + () -> resolveContactPointHostname(address.getHostString())); + pending.add(new PendingResolution(node, address, future)); + } catch (RejectedExecutionException e) { + // Executor already shut down (session closing): keep the unresolved node as a + // best-effort fallback rather than dropping it. + result.add(node); + } + continue; + } + } + // Already resolved or non-DefaultEndPoint endpoint — use as-is. + result.add(node); + } + + // Collect the concurrent resolutions against one shared deadline so the total wait is bounded + // by roughly a single timeout, no matter how many hostnames are pending. + long deadlineNanos = System.nanoTime() + getContactPointResolutionTimeout().toNanos(); + for (int i = 0; i < pending.size(); i++) { + PendingResolution p = pending.get(i); + InetAddress[] all; + try { + long remainingNanos = deadlineNanos - System.nanoTime(); + all = p.future.get(Math.max(0, remainingNanos), TimeUnit.NANOSECONDS); + } catch (TimeoutException e) { + p.future.cancel(true); + LOG.warn( + "[{}] Timed out resolving contact point hostname {} after {}, keeping it unresolved", + logPrefix, + p.address.getHostString(), + getContactPointResolutionTimeout()); + result.add(p.node); + continue; + } catch (ExecutionException e) { + LOG.warn( + "[{}] Could not resolve contact point hostname {}, keeping it unresolved", + logPrefix, + p.address.getHostString(), + e.getCause()); + result.add(p.node); + continue; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + // Keep this and every remaining contact point unresolved (best-effort) and stop waiting. + for (int j = i; j < pending.size(); j++) { + pending.get(j).future.cancel(true); + result.add(pending.get(j).node); + } + break; + } + if (all.length > 1) { + LOG.debug( + "[{}] Contact point {} expands to {} addresses", + logPrefix, + p.address.getHostString(), + all.length); + } + for (InetAddress ip : all) { + // Build the endpoint from the resolved InetAddress (not a raw IP string) on + // purpose: the InetAddress keeps the original hostname, so the TCP connection + // uses this IP while the SSL engine still receives the hostname for peer host, + // SNI and hostname verification. Do not simplify this to a numeric IP address. + InetSocketAddress resolved = new InetSocketAddress(ip, p.address.getPort()); + result.add(DefaultNode.newContactPoint(new DefaultEndPoint(resolved), context)); + } + } + return result; + } + + /** A contact-point hostname whose concurrent DNS resolution is in flight. */ + private static final class PendingResolution { + final DefaultNode node; + final InetSocketAddress address; + final Future future; + + PendingResolution(DefaultNode node, InetSocketAddress address, Future future) { + this.node = node; + this.address = address; + this.future = future; + } + } + + /** + * The timeout applied to each contact-point hostname resolution triggered by {@link + * #getResolvedContactPoints()}. Extracted as a method (rather than referencing the constant + * directly) so tests can substitute a short timeout instead of waiting out the real one. + */ + @VisibleForTesting + Duration getContactPointResolutionTimeout() { + return CONTACT_POINT_RESOLUTION_TIMEOUT; + } + + /** + * Resolves a contact-point hostname to all its IPs. Extracted as a method (rather than calling + * {@link InetAddress#getAllByName(String)} directly) so tests can substitute a slow or failing + * resolution to exercise the timeout path in {@link #getResolvedContactPoints()}. + */ + @VisibleForTesting + InetAddress[] resolveContactPointHostname(String host) throws UnknownHostException { + return InetAddress.getAllByName(host); + } + /** Whether the default contact point was used (because none were provided explicitly). */ public boolean wasImplicitContactPoint() { return wasImplicitContactPoint; @@ -188,6 +389,13 @@ public boolean wasImplicitContactPoint() { * they are never added to metadata and never exposed to user-facing APIs (events, {@link * com.datastax.oss.driver.api.core.metadata.Metadata#getNodes()}, or {@link * com.datastax.oss.driver.api.core.metadata.NodeStateListener} callbacks). + * + *

The metadata node stores {@code nodeInfo.getEndPoint()} as-is. When the control node was + * reached through a DNS-expanded synthetic contact point (see {@link + * #getResolvedContactPoints()}), that resolved IP endpoint is therefore persisted in metadata and + * is never re-resolved afterwards. Re-resolving the original hostname only happens through the + * original-contact-point reconnection fallback (see {@code + * advanced.control-connection.reconnection.fallback-to-original-contact-points}). */ public CompletionStage registerNode(NodeInfo nodeInfo) { Preconditions.checkNotNull(nodeInfo.getHostId(), "Cannot register node without hostId"); @@ -585,6 +793,7 @@ private void close() { if (queuedSchemaRefresh != null) { queuedSchemaRefresh.completeExceptionally(new IllegalStateException("Cluster is closed")); } + contactPointResolverExecutor.shutdownNow(); closeFuture.complete(null); } } diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/TopologyMonitor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/TopologyMonitor.java index 1bb8e343d96..420cd8e4f92 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/TopologyMonitor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/TopologyMonitor.java @@ -141,4 +141,24 @@ public interface TopologyMonitor extends AsyncAutoCloseable { * {@link DefaultTopologyMonitor}) should override this method. */ default void resetColumnCaches() {} + + /** + * Whether this monitor re-resolves node addresses dynamically on every connection attempt (for + * example by re-resolving a proxy hostname each time), rather than relying on an endpoint address + * captured once at node-registration time. + * + *

When this returns {@code true}, the control connection's reconnection query plan must not + * append the original contact points as a DNS re-resolution fallback (see {@code + * advanced.control-connection.reconnection.fallback-to-original-contact-points}): the monitor + * already keeps addresses fresh, and appending raw contact points could resurrect nodes that the + * monitor has authoritatively removed. + * + *

The default implementation returns {@code false}, which is correct for {@link + * DefaultTopologyMonitor}, whose {@code DefaultEndPoint}s cache their resolved address and never + * re-resolve. Proxy-based monitors that re-resolve per call should override this to return {@code + * true}. + */ + default boolean reresolvesNodeAddresses() { + return false; + } } diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf index 721afa1cd76..7240ec5252a 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1227,6 +1227,10 @@ datastax-java-driver { # Required: no (defaults to false) # Modifiable at runtime: no # Overridable in a profile: no + # + # DEPRECATED: this option no longer has any effect. Contact points are always kept as unresolved + # hostnames and expanded to all their DNS-mapped IPs at query-plan time (see + # MetadataManager.getResolvedContactPoints()). It will be removed in a future release. advanced.resolve-contact-points = false advanced.protocol { @@ -2318,14 +2322,20 @@ datastax-java-driver { } reconnection { - # Whether to forcibly add original contact points held by MetadataManager to the reconnection plan, - # in case there is no live nodes available according to LBP. - # Experimental. + # Whether to append the original contact points held by MetadataManager to the reconnection + # plan, after the live nodes reported by the load balancing policy. + # + # This is also the driver's DNS re-resolution path. Contact points are kept as unresolved + # hostnames and expanded to their current DNS IPs at query-plan time (see + # MetadataManager.getResolvedContactPoints()). Metadata nodes, in contrast, store an + # already-resolved endpoint that is never re-resolved, so once DNS records change they would + # otherwise become stale. Keeping this enabled lets control-connection reconnects re-resolve + # the original hostnames and pick up the new IPs once the live-node plan is exhausted. # # Required: yes # Modifiable at runtime: yes, the new value will be used for checks issued after the change. # Overridable in a profile: no - fallback-to-original-contact-points = false + fallback-to-original-contact-points = true } } diff --git a/core/src/test/java/com/datastax/dse/driver/internal/core/insights/InsightsClientTest.java b/core/src/test/java/com/datastax/dse/driver/internal/core/insights/InsightsClientTest.java index 5085432dbec..5f31090bf3f 100644 --- a/core/src/test/java/com/datastax/dse/driver/internal/core/insights/InsightsClientTest.java +++ b/core/src/test/java/com/datastax/dse/driver/internal/core/insights/InsightsClientTest.java @@ -487,7 +487,7 @@ private DefaultDriverContext mockDefaultDriverContext() throws UnknownHostExcept EndPoint contactEndPoint = mock(EndPoint.class); when(contactEndPoint.resolve()).thenReturn(new InetSocketAddress("127.0.0.1", 9999)); when(contactPoint.getEndPoint()).thenReturn(contactEndPoint); - when(manager.getContactPoints()).thenReturn(ImmutableSet.of(contactPoint)); + when(manager.getResolvedContactPoints()).thenReturn(ImmutableList.of(contactPoint)); DriverConfig driverConfig = mock(DriverConfig.class); when(context.getConfig()).thenReturn(driverConfig); diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitorTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitorTest.java index a1ba4617ef5..0a6268f3378 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitorTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitorTest.java @@ -28,6 +28,8 @@ import com.datastax.oss.driver.api.core.config.DriverConfig; import com.datastax.oss.driver.api.core.config.DriverExecutionProfile; import com.datastax.oss.driver.api.core.metadata.EndPoint; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; import com.datastax.oss.driver.internal.core.adminrequest.AdminResult; import com.datastax.oss.driver.internal.core.adminrequest.AdminRow; import com.datastax.oss.driver.internal.core.channel.DriverChannel; @@ -66,6 +68,8 @@ public class ClientRoutesTopologyMonitorTest { @Mock private ControlConnection controlConnection; @Mock private DriverConfig driverConfig; @Mock private DriverExecutionProfile defaultProfile; + @Mock private MetadataManager metadataManager; + @Mock private Metadata metadata; private TestableClientRoutesTopologyMonitor handler; @@ -220,6 +224,57 @@ public void should_refresh_updates_routes() throws UnknownHostException { assertThat(handler.resolve(hostId2)).isNotNull(); } + // ---- reresolvesNodeAddresses() ------------------------------------------- + + @Test + public void should_reresolve_when_all_known_nodes_have_client_routes() { + UUID hostId1 = UUID.randomUUID(); + UUID hostId2 = UUID.randomUUID(); + Node node1 = Mockito.mock(Node.class); + when(node1.getHostId()).thenReturn(hostId1); + Node node2 = Mockito.mock(Node.class); + when(node2.getHostId()).thenReturn(hostId2); + + when(context.getMetadataManager()).thenReturn(metadataManager); + when(metadataManager.getMetadata()).thenReturn(metadata); + when(metadata.getNodes()).thenReturn(ImmutableMap.of(hostId1, node1, hostId2, node2)); + + handler.setRoutes( + ImmutableMap.of( + hostId1, new ClientRouteRecord(hostId1, "127.0.0.1", 9042), + hostId2, new ClientRouteRecord(hostId2, "127.0.0.2", 9042))); + + assertThat(handler.reresolvesNodeAddresses()).isTrue(); + } + + @Test + public void should_not_reresolve_when_a_known_node_has_no_client_route() { + UUID hostId1 = UUID.randomUUID(); + UUID hostId2 = UUID.randomUUID(); + Node node1 = Mockito.mock(Node.class); + when(node1.getHostId()).thenReturn(hostId1); + Node node2 = Mockito.mock(Node.class); + when(node2.getHostId()).thenReturn(hostId2); + + when(context.getMetadataManager()).thenReturn(metadataManager); + when(metadataManager.getMetadata()).thenReturn(metadata); + when(metadata.getNodes()).thenReturn(ImmutableMap.of(hostId1, node1, hostId2, node2)); + + // Only node1 has a live client route; node2 would fall back to a static endpoint. + handler.setRoutes(ImmutableMap.of(hostId1, new ClientRouteRecord(hostId1, "127.0.0.1", 9042))); + + assertThat(handler.reresolvesNodeAddresses()).isFalse(); + } + + @Test + public void should_reresolve_when_no_nodes_known_yet() { + when(context.getMetadataManager()).thenReturn(metadataManager); + when(metadataManager.getMetadata()).thenReturn(metadata); + when(metadata.getNodes()).thenReturn(Collections.emptyMap()); + + assertThat(handler.reresolvesNodeAddresses()).isTrue(); + } + // ---- Merge behavior tests ----------------------------------------------- @Test diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapperTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapperTest.java index 89b36b9ee09..1865cc18576 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapperTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapperTest.java @@ -44,6 +44,7 @@ import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet; import com.datastax.oss.driver.shaded.guava.common.collect.Lists; +import java.util.ArrayList; import java.util.Map; import java.util.Objects; import java.util.Queue; @@ -79,6 +80,7 @@ public class LoadBalancingPolicyWrapperTest { private EventBus eventBus; @Mock private MetadataManager metadataManager; @Mock private Metadata metadata; + @Mock private TopologyMonitor topologyMonitor; @Mock protected MetricsFactory metricsFactory; @Captor private ArgumentCaptor> initNodesCaptor; @@ -100,8 +102,9 @@ public void setup() { Objects.requireNonNull(node3.getHostId()), node3); when(metadataManager.getMetadata()).thenReturn(metadata); when(metadata.getNodes()).thenReturn(allNodes); - when(metadataManager.getContactPoints()).thenReturn(contactPoints); + when(metadataManager.getResolvedContactPoints()).thenReturn(new ArrayList<>(contactPoints)); when(context.getMetadataManager()).thenReturn(metadataManager); + when(context.getTopologyMonitor()).thenReturn(topologyMonitor); when(context.getConfig()).thenReturn(config); when(config.getDefaultProfile()).thenReturn(defaultProfile); @@ -130,26 +133,36 @@ public void setup() { @Test public void should_build_control_connection_query_plan_from_contact_points_before_init() { + // Given — simulate DNS expansion: node1's hostname resolves to two IPs (node1 + node4) + DefaultNode node4 = TestNodeFactory.newNode(4, context); + when(metadataManager.getResolvedContactPoints()) + .thenReturn(ImmutableList.of(node1, node2, node4)); + // When Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); - // Then + // Then — query plan contains all DNS-expanded nodes, not just the original 2 contact points for (LoadBalancingPolicy policy : ImmutableList.of(policy1, policy2, policy3)) { verify(policy, never()).newQueryPlan(null, null); } - assertThat(queryPlan).hasSameElementsAs(contactPoints); + assertThat(queryPlan).containsExactlyInAnyOrder(node1, node2, node4); } @Test public void should_build_query_plan_from_contact_points_before_init() { + // Given — simulate DNS expansion: node1's hostname resolves to two IPs (node1 + node4) + DefaultNode node4 = TestNodeFactory.newNode(4, context); + when(metadataManager.getResolvedContactPoints()) + .thenReturn(ImmutableList.of(node1, node2, node4)); + // When Queue queryPlan = wrapper.newQueryPlan(null, DriverExecutionProfile.DEFAULT_NAME, null); - // Then + // Then — query plan contains all DNS-expanded nodes, not just the original 2 contact points for (LoadBalancingPolicy policy : ImmutableList.of(policy1, policy2, policy3)) { verify(policy, never()).newQueryPlan(null, null); } - assertThat(queryPlan).hasSameElementsAs(contactPoints); + assertThat(queryPlan).containsExactlyInAnyOrder(node1, node2, node4); } @Test @@ -204,8 +217,7 @@ public void should_fetch_control_connection_query_plan_from_policy_after_init() assertThat(queryPlan.poll()).isEqualTo(node3); assertThat(queryPlan.poll()).isEqualTo(node2); assertThat(queryPlan.poll()).isEqualTo(node1); - // Remaining nodes are contact points appended at the end. - // They are new DefaultNode instances created via newContactPoint, so compare by endpoint. + // Remaining nodes are the DNS-expanded (resolved) contact points appended at the end. Set remainingEndpoints = new java.util.HashSet<>(); for (Node n : queryPlan) { remainingEndpoints.add(n.getEndPoint()); @@ -217,6 +229,56 @@ public void should_fetch_control_connection_query_plan_from_policy_after_init() assertThat(remainingEndpoints).isEqualTo(contactEndpoints); } + @Test + public void should_not_double_resolve_contact_points_before_init() { + // Given — the wrapper hasn't been init()-ed yet (state=BEFORE_INIT), so newQueryPlan() already + // builds the regular plan directly from resolved contact points. The reconnect-contact-points + // flag doesn't matter here: newControlReconnectionQueryPlan() short-circuits on state before + // even reading it, since appending contact points again pre-init would just duplicate every + // entry in the plan and re-trigger DNS resolution for no benefit. + + // When + Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); + + // Then — getResolvedContactPoints() is resolved only once (not once for the regular plan and + // again for a redundant "fallback" append), and the plan has no duplicate entries. + verify(metadataManager, times(1)).getResolvedContactPoints(); + assertThat(queryPlan).containsExactlyInAnyOrder(node1, node2); + } + + @Test + public void + should_not_append_contact_points_to_query_plan_when_reconnect_contact_points_is_disabled() { + // Given — the flag defaults to false in the test setup (see @Before) + wrapper.init(); + + // When + Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); + + // Then + // Only the policy query plan is returned; no contact points are appended. + assertThat(queryPlan).isEqualTo(defaultPolicyQueryPlan); + } + + @Test + public void + should_not_append_contact_points_to_query_plan_when_topology_monitor_reresolves_addresses() { + // Given — the flag is enabled, but the topology monitor re-resolves node addresses on its own + // (e.g. a proxy-based monitor such as client routes or the cloud SNI proxy). + when(defaultProfile.getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS)) + .thenReturn(true); + when(topologyMonitor.reresolvesNodeAddresses()).thenReturn(true); + wrapper.init(); + + // When + Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); + + // Then + // Contact points must not be appended: the monitor keeps addresses fresh, and appending raw + // contact points could resurrect nodes it has authoritatively removed. + assertThat(queryPlan).isEqualTo(defaultPolicyQueryPlan); + } + @Test public void should_return_contact_points_when_query_plan_empty_and_flag_enabled() { // Given diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/MetadataManagerTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/MetadataManagerTest.java index 6fb1384d889..2d3163b32c1 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/MetadataManagerTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/MetadataManagerTest.java @@ -42,7 +42,9 @@ import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet; import io.netty.channel.DefaultEventLoopGroup; +import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.UnknownHostException; import java.time.Duration; import java.util.Collections; import java.util.List; @@ -490,8 +492,154 @@ public void should_throw_on_registerNode_with_null_hostId() { .hasMessageContaining("Cannot register node without hostId"); } + @Test + public void should_return_empty_list_when_contact_points_not_yet_set() { + // contactPoints field is null until addContactPoints is called + assertThat(metadataManager.getResolvedContactPoints()).isEmpty(); + } + + @Test + public void should_return_already_resolved_contact_points_unchanged() { + // Given — a contact point with an already-resolved InetSocketAddress + metadataManager.addContactPoints(ImmutableSet.of(END_POINT2)); + + // When + List resolved = metadataManager.getResolvedContactPoints(); + + // Then — the single node is returned as-is (no expansion needed) + assertThat(resolved).hasSize(1); + assertThat(resolved.get(0).getEndPoint()).isEqualTo(END_POINT2); + } + + @Test + public void should_expand_unresolved_hostname_to_all_ips() { + // Given — a contact point with an unresolved hostname (localhost → 127.0.0.1) + EndPoint unresolvedEndPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("localhost", 9042)); + metadataManager.addContactPoints(ImmutableSet.of(unresolvedEndPoint)); + + // When + List resolved = metadataManager.getResolvedContactPoints(); + + // Then — at least one node is returned, each with a resolved address + assertThat(resolved).isNotEmpty(); + for (Node node : resolved) { + InetSocketAddress addr = (InetSocketAddress) node.getEndPoint().resolve(); + assertThat(addr.isUnresolved()).isFalse(); + assertThat(addr.getPort()).isEqualTo(9042); + } + } + + @Test + public void should_expand_multiple_contact_points_independently() { + // Given — two contact points: one already resolved, one unresolved + EndPoint resolvedEndPoint = END_POINT3; + EndPoint unresolvedEndPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("localhost", 9042)); + metadataManager.addContactPoints(ImmutableSet.of(resolvedEndPoint, unresolvedEndPoint)); + + // When + List resolved = metadataManager.getResolvedContactPoints(); + + // Then — at least 2 nodes: 1 for the resolved + at least 1 for localhost expansion + assertThat(resolved.size()).isGreaterThanOrEqualTo(2); + // The resolved endpoint must appear + assertThat(resolved).anySatisfy(n -> assertThat(n.getEndPoint()).isEqualTo(resolvedEndPoint)); + // All returned addresses must be resolved + for (Node node : resolved) { + InetSocketAddress addr = (InetSocketAddress) node.getEndPoint().resolve(); + assertThat(addr.isUnresolved()).isFalse(); + } + } + + @Test + public void should_keep_unresolved_contact_point_when_hostname_cannot_be_resolved() { + // Given — a hostname that is guaranteed to fail DNS resolution + EndPoint badEndPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("nonexistent.invalid", 9042)); + metadataManager.addContactPoints(ImmutableSet.of(badEndPoint)); + + // When + List resolved = metadataManager.getResolvedContactPoints(); + + // Then — resolution is best-effort: the unresolvable hostname is kept as-is (logged as WARN) + // instead of being dropped, so the query plan is never emptier than the configured contact + // points and the hostname can still be resolved later at connection time. + assertThat(resolved).hasSize(1); + assertThat(resolved.get(0).getEndPoint()).isEqualTo(badEndPoint); + InetSocketAddress addr = (InetSocketAddress) resolved.get(0).getEndPoint().resolve(); + assertThat(addr.isUnresolved()).isTrue(); + } + + @Test + public void should_keep_unresolved_contact_point_when_resolution_times_out() { + // Given — a hostname whose resolution is artificially delayed well past the (test-shortened) + // resolution timeout + EndPoint slowEndPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("slow.invalid", 9042)); + metadataManager.addContactPoints(ImmutableSet.of(slowEndPoint)); + metadataManager.delayResolutionOf("slow.invalid"); + + // When + long startNanos = System.nanoTime(); + List resolved = metadataManager.getResolvedContactPoints(); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + + // Then — the call returns bounded by roughly the (test) timeout instead of hanging for the full + // simulated delay, and the timed-out hostname is kept unresolved (best-effort) like an + // unresolvable one, not dropped. + assertThat(resolved).hasSize(1); + assertThat(resolved.get(0).getEndPoint()).isEqualTo(slowEndPoint); + InetSocketAddress addr = (InetSocketAddress) resolved.get(0).getEndPoint().resolve(); + assertThat(addr.isUnresolved()).isTrue(); + assertThat(elapsedMillis).isLessThan(1500); + } + + @Test + public void should_not_let_one_slow_hostname_starve_the_others() { + // Given — one hostname whose resolution hangs past the timeout, alongside a healthy one. + // Resolutions run concurrently, so the slow one must not prevent the healthy one from resolving + // (the poisoning a single shared resolver thread would cause). + EndPoint slowEndPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("slow.invalid", 9042)); + EndPoint healthyEndPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("localhost", 9042)); + metadataManager.addContactPoints(ImmutableSet.of(slowEndPoint, healthyEndPoint)); + metadataManager.delayResolutionOf("slow.invalid"); + + // When + long startNanos = System.nanoTime(); + List resolved = metadataManager.getResolvedContactPoints(); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + + // Then — the whole call stays bounded by ~one timeout, the slow hostname is kept unresolved, + // and the healthy hostname still resolves to a concrete IP (it was not starved behind the slow + // one). + assertThat(elapsedMillis).isLessThan(1500); + assertThat(resolved) + .anySatisfy( + n -> { + InetSocketAddress a = (InetSocketAddress) n.getEndPoint().resolve(); + assertThat(a.getHostString()).isEqualTo("slow.invalid"); + assertThat(a.isUnresolved()).isTrue(); + }); + assertThat(resolved) + .anySatisfy( + n -> { + InetSocketAddress a = (InetSocketAddress) n.getEndPoint().resolve(); + assertThat(a.isUnresolved()).isFalse(); + assertThat(a.getPort()).isEqualTo(9042); + }); + } + private static class TestMetadataManager extends MetadataManager { + // Shortened so should_skip_contact_point_when_resolution_times_out() doesn't have to wait out + // the real (3s) production timeout. + private static final Duration TEST_RESOLUTION_TIMEOUT = Duration.ofMillis(200); + + private volatile String delayedHostname; + private List refreshes = new CopyOnWriteArrayList<>(); private volatile int addNodeCount = 0; private volatile int removeNodeCount = 0; @@ -500,6 +648,28 @@ public TestMetadataManager(InternalDriverContext context) { super(context); } + void delayResolutionOf(String hostname) { + this.delayedHostname = hostname; + } + + @Override + Duration getContactPointResolutionTimeout() { + return TEST_RESOLUTION_TIMEOUT; + } + + @Override + InetAddress[] resolveContactPointHostname(String host) throws UnknownHostException { + if (host.equals(delayedHostname)) { + try { + // Sleep well past TEST_RESOLUTION_TIMEOUT so the caller's Future.get() times out first. + Thread.sleep(TEST_RESOLUTION_TIMEOUT.toMillis() * 10); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + return super.resolveContactPointHostname(host); + } + @Override Void apply(MetadataRefresh refresh) { // Do not execute refreshes, just store them for inspection in the test diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/heartbeat/HeartbeatIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/heartbeat/HeartbeatIT.java index 26658bd76d1..b8a32c68450 100644 --- a/integration-tests/src/test/java/com/datastax/oss/driver/core/heartbeat/HeartbeatIT.java +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/heartbeat/HeartbeatIT.java @@ -235,6 +235,10 @@ private CqlSession newSession(ProgrammaticDriverConfigLoaderBuilder loaderBuilde .withDuration(DefaultDriverOption.HEARTBEAT_TIMEOUT, Duration.ofMillis(500)) .withDuration(DefaultDriverOption.CONNECTION_INIT_QUERY_TIMEOUT, Duration.ofSeconds(2)) .withDuration(DefaultDriverOption.RECONNECTION_MAX_DELAY, Duration.ofSeconds(1)) + // These tests exercise heartbeat behavior only. Disable the contact-point + // reconnection fallback, which would otherwise send an extra OPTIONS message on + // init/reconnect and skew the heartbeat counts. + .withBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, false) .build(); return SessionUtils.newSession(SIMULACRON_RULE, loader); } diff --git a/integration-tests/src/test/java/com/datastax/oss/driver/core/resolver/MockResolverIT.java b/integration-tests/src/test/java/com/datastax/oss/driver/core/resolver/MockResolverIT.java index 4e9eefebf63..c8443cfe90b 100644 --- a/integration-tests/src/test/java/com/datastax/oss/driver/core/resolver/MockResolverIT.java +++ b/integration-tests/src/test/java/com/datastax/oss/driver/core/resolver/MockResolverIT.java @@ -25,7 +25,6 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import com.datastax.oss.driver.api.core.CqlSession; @@ -109,7 +108,46 @@ public void should_connect_with_mocked_hostname() { assertThat(filteredNodes).hasSize(1); InetSocketAddress address = (InetSocketAddress) filteredNodes.iterator().next().getEndPoint().resolve(); - assertTrue(address.isUnresolved()); + assertFalse(address.isUnresolved()); + } + } + } + + @Test + public void should_connect_when_first_dns_entry_is_non_responsive() { + final int numberOfNodes = 2; + DriverConfigLoader loader = + new DefaultProgrammaticDriverConfigLoaderBuilder() + .withBoolean(TypedDriverOption.RESOLVE_CONTACT_POINTS.getRawOption(), false) + .withBoolean(TypedDriverOption.RECONNECT_ON_INIT.getRawOption(), true) + .withStringList( + TypedDriverOption.CONTACT_POINTS.getRawOption(), + Collections.singletonList("test.cluster.fake:9042")) + .build(); + + CqlSessionBuilder builder = new CqlSessionBuilder().withConfigLoader(loader); + try (CcmBridge ccmBridge = + CcmBridge.builder().withNodes(numberOfNodes).withIpPrefix("127.0.1.").build()) { + MultimapHostResolverProvider.removeResolverEntries("test.cluster.fake"); + // Register the dead IP first so it's the first entry InetAddress.getAllByName() returns for + // this hostname (MultimapHostResolver preserves insertion order). Node 11 is never started, + // so nothing listens on 127.0.1.11 in this subnet. + MultimapHostResolverProvider.addResolverEntry("test.cluster.fake", "127.0.1.11"); + MultimapHostResolverProvider.addResolverEntry( + "test.cluster.fake", ccmBridge.getNodeIpAddress(1)); + MultimapHostResolverProvider.addResolverEntry( + "test.cluster.fake", ccmBridge.getNodeIpAddress(2)); + ccmBridge.create(); + ccmBridge.start(); + + try (CqlSession session = builder.build()) { + waitForAllNodesUp(session, numberOfNodes); + ResultSet rs = session.execute("select * from system.local where key='local'"); + assertThat(rs).isNotNull(); + List rows = rs.all(); + assertThat(rows).hasSize(1); + Collection nodes = session.getMetadata().getNodes().values(); + assertThat(nodes).hasSize(numberOfNodes); } } }