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); } } }