diff --git a/core/src/main/java/com/datastax/dse/driver/api/core/auth/DseGssApiAuthProviderBase.java b/core/src/main/java/com/datastax/dse/driver/api/core/auth/DseGssApiAuthProviderBase.java index 48a0e5b0ef3..593552d77c1 100644 --- a/core/src/main/java/com/datastax/dse/driver/api/core/auth/DseGssApiAuthProviderBase.java +++ b/core/src/main/java/com/datastax/dse/driver/api/core/auth/DseGssApiAuthProviderBase.java @@ -291,6 +291,8 @@ protected static class GssApiAuthenticator extends BaseDseAuthenticator { private SaslClient saslClient; private EndPoint endPoint; + @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); + // Kerberos authentication needs a single canonical hostname for SASL service name resolution. protected GssApiAuthenticator( GssApiOptions options, EndPoint endPoint, String serverAuthenticator) { super(serverAuthenticator); 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..54d723152a4 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 @@ -288,6 +288,8 @@ private InsightsStatusData createStatusData() { .build(); } + @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); + // address reporting needs a single canonical address per node. private Map getConnectedNodes() { Map pools = driverContext.getPoolManager().getPools(); return pools.entrySet().stream() @@ -302,6 +304,8 @@ private SessionStateForNode constructSessionStateForNode(Map.Entry startupOptions = driverContext.getStartupOptions(); return InsightsStartupData.builder() @@ -454,6 +458,8 @@ private PoolSizeByHostDistance getPoolSizeByHostDistance() { 0); } + @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); + // address reporting needs a single canonical address for the control connection. private String getControlConnectionSocketAddress() { SocketAddress controlConnectionAddress = controlConnection.channel().getEndPoint().resolve(); return AddressFormatter.nullSafeToString(controlConnectionAddress); 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..d674662644e 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,15 @@ 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 control-connection reconnection plan, + * after the live nodes reported by the load balancing policy (defaults to {@code true}). + * + *

Contact points are appended as-is (unresolved hostnames); each is expanded to all of its + * current DNS IPs at connection time via {@code EndPoint.resolveAll()}, which is also the + * driver's DNS re-resolution mechanism. The append is skipped for topology monitors that + * re-resolve node addresses themselves (such as the cloud/proxy monitors). + */ public static final TypedDriverOption CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS = new TypedDriverOption<>( DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, GenericType.BOOLEAN); @@ -664,7 +672,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/metadata/EndPoint.java b/core/src/main/java/com/datastax/oss/driver/api/core/metadata/EndPoint.java index 530f2ad38ac..b4c6a03286b 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/metadata/EndPoint.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/metadata/EndPoint.java @@ -18,28 +18,75 @@ package com.datastax.oss.driver.api.core.metadata; import edu.umd.cs.findbugs.annotations.NonNull; -import java.net.InetSocketAddress; import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; /** * Encapsulates the information needed to open connections to a node. * *

By default, the driver assumes plain TCP connections, and this is just a wrapper around an - * {@link InetSocketAddress}. However, more complex deployment scenarios might use a custom + * {@link java.net.InetSocketAddress}. However, more complex deployment scenarios might use a custom * implementation that contains additional information; for example, if the nodes are accessed * through a proxy with SNI routing, an SNI server name is needed in addition to the proxy address. */ public interface EndPoint { /** - * Resolves this instance to a socket address. + * Resolves this instance to a single socket address. * *

This will be called each time the driver opens a new connection to the node. The returned * address cannot be null. + * + * @deprecated Use {@link #resolveAll(Executor)} instead. When a hostname maps to multiple IPs + * (e.g. in dynamic DNS environments) only one address is returned here, causing the driver to + * miss fallback IPs when the first one is unreachable. {@code resolveAll(Executor)} returns + * the full set, resolved asynchronously off the calling thread. */ + @Deprecated @NonNull SocketAddress resolve(); + /** + * Resolves this instance to all known socket addresses, asynchronously. + * + *

This is called each time the driver opens a new connection to the node. For endpoints backed + * by a plain IP address the returned array contains exactly one element. For endpoints whose + * hostname resolves to multiple IPs (e.g. a DNS round-robin entry) all addresses are returned so + * that the driver can try each one in sequence and fall back gracefully when individual IPs are + * unreachable. + * + *

Resolution is asynchronous on purpose: name resolution can block (e.g. {@link + * java.net.InetAddress#getAllByName(String)}), and the driver calls this from its admin event + * loop, which must never block. Implementations whose resolution may block must run it on + * the supplied {@code executor} rather than on the calling thread (see the default + * implementation). Implementations that resolve from memory (e.g. an already-resolved address or + * an in-memory lookup) may return an {@linkplain CompletableFuture#completedFuture(Object) + * already completed stage} and ignore the executor. + * + *

The default implementation offloads {@link #resolve()} to {@code executor} and wraps the + * result in a single-element array. Implementations that can supply multiple addresses should + * override this method. + * + *

The returned stage must not be null and must complete with a non-null, non-empty array. + * + * @param executor the executor to run potentially-blocking resolution on; must not be null. The + * driver supplies a dedicated resolver executor so that blocking name resolution never runs + * on an event loop. + * @apiNote Timeout note: {@link + * com.datastax.oss.driver.internal.core.channel.ChannelFactory} tries each address in + * sequence. If a hostname resolves to N addresses and each attempt times out, the worst-case + * connection time before declaring a node unreachable is {@code N × + * advanced.connection.connect-timeout}. In practice DNS round-robin entries have only a small + * number of records, so this is rarely a concern, but callers should be aware of this when + * configuring connect timeouts. + */ + @NonNull + default CompletionStage resolveAll(@NonNull Executor executor) { + return CompletableFuture.supplyAsync(() -> new SocketAddress[] {resolve()}, executor); + } + /** * Returns an alternate string representation for use in node-level metric names. * 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..85a2e0488b3 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 connection time (via {@code EndPoint.resolveAll()}), 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,10 @@ 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 connection time via EndPoint.resolveAll(). 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/channel/ChannelFactory.java b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelFactory.java index 6bfc355f910..c0e6276343b 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelFactory.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/channel/ChannelFactory.java @@ -58,12 +58,16 @@ import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.SocketAddress; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Optional; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; import java.util.concurrent.CompletionStage; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import net.jcip.annotations.ThreadSafe; @@ -107,6 +111,12 @@ public class ChannelFactory { private final String logPrefix; protected final InternalDriverContext context; + // Runs potentially-blocking EndPoint name resolution (EndPoint.resolveAll()) off the caller + // thread: connect() is invoked from the admin event loop, which must never block on DNS. A cached + // pool gives each concurrent resolution its own daemon thread, so a single blackholed lookup + // cannot starve other nodes' reconnections; idle threads are reclaimed after ~60s. + private final ExecutorService resolverExecutor; + /** either set from the configuration, or null and will be negotiated */ @VisibleForTesting volatile ProtocolVersion protocolVersion; @@ -124,6 +134,18 @@ public ChannelFactory(InternalDriverContext context) { this.logPrefix = context.getSessionName(); this.context = context; + AtomicInteger resolverThreadCount = new AtomicInteger(); + this.resolverExecutor = + Executors.newCachedThreadPool( + runnable -> { + Thread thread = + new Thread( + runnable, + logPrefix + "-connection-resolver-" + resolverThreadCount.incrementAndGet()); + thread.setDaemon(true); + return thread; + }); + DriverExecutionProfile defaultConfig = context.getConfig().getDefaultProfile(); if (defaultConfig.isDefined(DefaultDriverOption.PROTOCOL_VERSION)) { String versionName = defaultConfig.getString(DefaultDriverOption.PROTOCOL_VERSION); @@ -154,6 +176,14 @@ public String getClusterName() { return clusterName; } + /** + * Shuts down the internal DNS-resolver executor. Invoked during session shutdown. The executor's + * threads are daemon threads, so a missed call cannot prevent JVM exit. + */ + public void close() { + resolverExecutor.shutdownNow(); + } + public CompletionStage connect(Node node, DriverChannelOptions options) { NodeMetricUpdater nodeMetricUpdater; if (node instanceof DefaultNode) { @@ -219,14 +249,153 @@ private void connect( List attemptedVersions, CompletableFuture resultFuture) { - SocketAddress resolvedAddress; + // Resolution may block (DNS); it is offloaded to resolverExecutor so the calling thread (the + // admin event loop, for control-connection reconnects) never waits on it. The continuation + // below runs off the admin loop -- tryNextCandidate()/Bootstrap.connect() are safe from any + // thread and per-candidate retries already run on Netty I/O threads. + CompletionStage resolveStage; try { - resolvedAddress = endPoint.resolve(); + resolveStage = endPoint.resolveAll(resolverExecutor); } catch (Exception e) { + // An implementation may throw synchronously while building the stage. resultFuture.completeExceptionally(e); return; } + resolveStage.whenComplete( + (candidates, error) -> { + if (error != null) { + // supplyAsync wraps supplier throwables in a CompletionException; unwrap so callers + // (and the guard test's isSameAs assertion) see the original cause. + Throwable cause = + (error instanceof CompletionException && error.getCause() != null) + ? error.getCause() + : error; + resultFuture.completeExceptionally(cause); + return; + } + if (candidates == null || candidates.length == 0) { + resultFuture.completeExceptionally( + new IllegalArgumentException( + "EndPoint.resolveAll() must return a non-null, non-empty array: " + endPoint)); + return; + } + tryNextCandidate( + endPoint, + shardingInfo, + shardId, + options, + nodeMetricUpdater, + currentVersion, + isNegotiating, + attemptedVersions, + resultFuture, + candidates, + 0, + new ArrayList<>()); + }); + } + + /** + * Iterates through the candidate addresses from {@link + * EndPoint#resolveAll(java.util.concurrent.Executor)}. Tries each one in sequence; if an address + * fails for a reason other than protocol-version negotiation exhaustion, the next candidate is + * tried. Only when all candidates are exhausted is the overall {@code resultFuture} failed. + * + *

Timeout note: addresses are tried serially, so the worst-case time before failure is + * {@code N × connect-timeout} where N is the number of candidates. This is an intentional + * tradeoff: failing immediately on the first unreachable IP would prevent fallback to healthy + * ones. In practice DNS entries have only a small number of records. + * + *

When every candidate fails, the last candidate's error is propagated with each earlier + * candidate's failure attached as a {@linkplain Throwable#addSuppressed(Throwable) suppressed} + * exception, so the full set of per-address causes is visible for diagnosis (they are otherwise + * only logged at DEBUG). + */ + private void tryNextCandidate( + EndPoint endPoint, + NodeShardingInfo shardingInfo, + Integer shardId, + DriverChannelOptions options, + NodeMetricUpdater nodeMetricUpdater, + ProtocolVersion currentVersion, + boolean isNegotiating, + List attemptedVersions, + CompletableFuture resultFuture, + SocketAddress[] candidates, + int index, + List priorErrors) { + + SocketAddress candidate = candidates[index]; + CompletableFuture perAddressFuture = new CompletableFuture<>(); + connectToAddress( + endPoint, + shardingInfo, + shardId, + options, + nodeMetricUpdater, + currentVersion, + isNegotiating, + attemptedVersions, + perAddressFuture, + candidate); + + perAddressFuture.whenComplete( + (channel, error) -> { + if (error == null) { + resultFuture.complete(channel); + } else if (index + 1 < candidates.length) { + LOG.debug( + "[{}] Failed to connect to {} ({}), trying next address", + logPrefix, + candidate, + error.getMessage()); + priorErrors.add(error); + tryNextCandidate( + endPoint, + shardingInfo, + shardId, + options, + nodeMetricUpdater, + currentVersion, + isNegotiating, + attemptedVersions, + resultFuture, + candidates, + index + 1, + priorErrors); + } else { + // All candidates exhausted. Surface the last error, carrying the earlier failures as + // suppressed exceptions so they are not lost (they were only logged at DEBUG above). + for (Throwable priorError : priorErrors) { + if (priorError != error) { + error.addSuppressed(priorError); + } + } + // Note: might be completed already if the failure happened in initializer() + resultFuture.completeExceptionally(error); + } + }); + } + + /** + * Performs a Netty bootstrap connect to a single, already-resolved address. Handles + * protocol-version negotiation (downgrade retries) internally, staying on the same address. Uses + * {@code perAddressFuture} so {@link #tryNextCandidate} can distinguish a per-address TCP failure + * (try the next IP) from a successful protocol handshake. + */ + private void connectToAddress( + EndPoint endPoint, + NodeShardingInfo shardingInfo, + Integer shardId, + DriverChannelOptions options, + NodeMetricUpdater nodeMetricUpdater, + ProtocolVersion currentVersion, + boolean isNegotiating, + List attemptedVersions, + CompletableFuture perAddressFuture, + SocketAddress resolvedAddress) { + NettyOptions nettyOptions = context.getNettyOptions(); Bootstrap bootstrap = @@ -235,7 +404,8 @@ private void connect( .channel(nettyOptions.channelClass()) .option(ChannelOption.ALLOCATOR, nettyOptions.allocator()) .handler( - initializer(endPoint, currentVersion, options, nodeMetricUpdater, resultFuture)); + initializer( + endPoint, currentVersion, options, nodeMetricUpdater, perAddressFuture)); nettyOptions.afterBootstrapInitialized(bootstrap); @@ -294,7 +464,7 @@ private void connect( ConsistencyLevel.LOCAL_QUORUM.name())); } } - resultFuture.complete(driverChannel); + perAddressFuture.complete(driverChannel); } else { Throwable error = connectFuture.cause(); if (error instanceof UnsupportedProtocolVersionException && isNegotiating) { @@ -307,7 +477,8 @@ private void connect( logPrefix, currentVersion, downgraded.get()); - connect( + // Stay on the same address for protocol-version downgrade retries. + connectToAddress( endPoint, shardingInfo, shardId, @@ -316,16 +487,17 @@ private void connect( downgraded.get(), true, attemptedVersions, - resultFuture); + perAddressFuture, + resolvedAddress); } else { - resultFuture.completeExceptionally( + perAddressFuture.completeExceptionally( UnsupportedProtocolVersionException.forNegotiation( endPoint, attemptedVersions)); } } else { // Note: might be completed already if the failure happened in initializer(), this is // fine - resultFuture.completeExceptionally(error); + perAddressFuture.completeExceptionally(error); } } }); 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/ClientRoutesEndPoint.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesEndPoint.java index 15d825b2efc..62976f08cb3 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesEndPoint.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesEndPoint.java @@ -27,6 +27,9 @@ import java.net.SocketAddress; import java.util.Objects; import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; public class ClientRoutesEndPoint implements EndPoint { private final UUID hostId; @@ -76,6 +79,21 @@ public SocketAddress resolve() { return fallbackEndPoint.resolve(); } + /** + * Returns all socket addresses for this endpoint. + * + *

Delegates to {@link #resolve()} to obtain the single address provided by the topology + * monitor (or the fallback endpoint), then returns it as a one-element array in an already + * completed stage. The topology monitor resolves each node to exactly one address by design (via + * an in-memory per-host-id lookup), so multi-address expansion is not applicable here and the + * {@code executor} is not used. + */ + @NonNull + @Override + public CompletionStage resolveAll(@NonNull Executor executor) { + return CompletableFuture.completedFuture(new SocketAddress[] {resolve()}); + } + @Override public boolean equals(Object other) { if (other == this) { 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/DefaultEndPoint.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPoint.java index 7ffbee8e4bb..a7bec899ccd 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPoint.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPoint.java @@ -20,8 +20,14 @@ import com.datastax.oss.driver.api.core.metadata.EndPoint; import edu.umd.cs.findbugs.annotations.NonNull; import java.io.Serializable; +import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.UnknownHostException; import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; public class DefaultEndPoint implements EndPoint, Serializable { @@ -41,6 +47,50 @@ public InetSocketAddress resolve() { return address; } + /** + * Returns all socket addresses for this endpoint. + * + *

If the stored address is already resolved, the returned stage completes immediately (on the + * calling thread) with a single-element array; no executor hop occurs. + * + *

If the stored address is unresolved (i.e. the driver was configured with {@code + * RESOLVE_CONTACT_POINTS=false} and the hostname has not been looked up yet), the blocking {@link + * InetAddress#getAllByName(String)} lookup is run on {@code executor} to expand the hostname to + * every IP it resolves to. Each resolved IP is returned as an {@link InetSocketAddress} with the + * same port as the original. + * + *

If DNS resolution fails, falls back to a single-element array containing {@link #resolve()}. + * + *

Note on resolver: DNS lookup is performed via {@link + * InetAddress#getAllByName(String)}, bypassing any custom Netty {@code AddressResolverGroup} + * configured on the driver. This is consistent with how {@link SniEndPoint} performs DNS + * resolution elsewhere in the driver. Users who rely on a custom Netty resolver should supply + * pre-resolved {@link java.net.InetSocketAddress} instances instead of hostnames. + */ + @NonNull + @Override + public CompletionStage resolveAll(@NonNull Executor executor) { + if (!address.isUnresolved()) { + return CompletableFuture.completedFuture(new SocketAddress[] {address}); + } + return CompletableFuture.supplyAsync( + () -> { + try { + InetAddress[] all = InetAddress.getAllByName(address.getHostString()); + SocketAddress[] result = new SocketAddress[all.length]; + for (int i = 0; i < all.length; i++) { + result[i] = new InetSocketAddress(all[i], address.getPort()); + } + return result; + } catch (UnknownHostException e) { + // Fallback: return the single unresolved address; the connect attempt will fail with a + // descriptive error rather than silently returning an empty array. + return new SocketAddress[] {address}; + } + }, + executor); + } + @Override public boolean equals(Object other) { if (other == this) { diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultTopologyMonitor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultTopologyMonitor.java index 5a82bfe2c86..2098a3e6553 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultTopologyMonitor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/DefaultTopologyMonitor.java @@ -701,6 +701,8 @@ private Optional findInPeers( // Current versions of Cassandra (3.11 at the time of writing), require the same port for all // nodes. As a consequence, the port is not stored in system tables. // We save it the first time we get a control connection channel. + @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); a single + // canonical address is all that is needed here to extract the port. protected void savePort(DriverChannel channel) { if (port < 0) { SocketAddress address = channel.getEndPoint().resolve(); @@ -723,6 +725,8 @@ protected void savePort(DriverChannel channel) { * otherwise. */ @Nullable + @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); a single + // canonical address is all that is needed here for the peer-vs-local comparison. protected InetSocketAddress getBroadcastRpcAddress( @NonNull AdminRow row, @NonNull EndPoint localEndPoint) { 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..fc1390f4ac2 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 @@ -27,6 +27,8 @@ import com.datastax.oss.driver.api.core.session.Request; import com.datastax.oss.driver.api.core.session.Session; import com.datastax.oss.driver.internal.core.context.InternalDriverContext; +import com.datastax.oss.driver.internal.core.util.collection.CompositeQueryPlan; +import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan; import com.datastax.oss.driver.internal.core.util.concurrent.ReplayingEventFilter; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableMap; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableSet; @@ -147,7 +149,9 @@ public Queue newQueryPlan( switch (stateRef.get()) { case BEFORE_INIT: case DURING_INIT: - // The contact points are not stored in the metadata yet: + // The contact points are not stored in the metadata yet. Each unresolved hostname is + // expanded to all its DNS IPs at connection time by EndPoint.resolveAll(), so one entry per + // contact point is enough here. List nodes = new ArrayList<>(context.getMetadataManager().getContactPoints()); Collections.shuffle(nodes); return new ConcurrentLinkedQueue<>(nodes); @@ -164,20 +168,42 @@ 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 the + // 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 the contact points as an explicit fallback once the LBP is RUNNING: before that + // (BEFORE_INIT/DURING_INIT), newQueryPlan() above already built regularQueryPlan directly from + // the contact points, so appending them again here would just duplicate every entry. + // + // 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. The exception is an empty regular plan: with no live node to try, + // reconnection cannot recover on its own, so the contact-point fallback is kept even for those + // monitors. + if (state == State.RUNNING + && context + .getConfig() + .getDefaultProfile() + .getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS) + && (!context.getTopologyMonitor().reresolvesNodeAddresses() + || regularQueryPlan.isEmpty())) { + // Append the original (unresolved) contact points so every IP their hostname resolves to is + // tried as a fallback: EndPoint.resolveAll() expands each one at connection time, instead of + // the driver being stuck with whatever single IP a metadata node happens to hold. List contactNodes = new ArrayList<>(); - for (DefaultNode node : originalNodes) { + for (DefaultNode node : context.getMetadataManager().getContactPoints()) { contactNodes.add(DefaultNode.newContactPoint(node.getEndPoint(), context)); } Collections.shuffle(contactNodes); - // Append contact points to the end of the regular query plan so they serve as a fallback - regularQueryPlan.addAll(contactNodes); + // Concatenate rather than mutate: the RUNNING-state regularQueryPlan is a built-in QueryPlan + // whose add()/addAll() throw UnsupportedOperationException (poll() is its only mutator). + // CompositeQueryPlan drains the regular plan first, then the contact-point fallback. + return new CompositeQueryPlan(regularQueryPlan, new SimpleQueryPlan(contactNodes.toArray())); } return regularQueryPlan; 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..2ca9b6f1012 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 @@ -188,6 +188,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 and never re-resolves it on + * its own. Re-resolving the original contact-point hostname to pick up current DNS only happens + * through the original-contact-point reconnection fallback (see {@code + * advanced.control-connection.reconnection.fallback-to-original-contact-points}), which re-enters + * the contact points and lets {@link EndPoint#resolveAll(java.util.concurrent.Executor)} expand + * each hostname at connection time. */ public CompletionStage registerNode(NodeInfo nodeInfo) { Preconditions.checkNotNull(nodeInfo.getHostId(), "Cannot register node without hostId"); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java index d1ab8eec98d..a141ef1576b 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java @@ -22,14 +22,25 @@ import edu.umd.cs.findbugs.annotations.NonNull; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.SocketAddress; import java.net.UnknownHostException; import java.util.Arrays; import java.util.Comparator; import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicInteger; public class SniEndPoint implements EndPoint { + // Rotates the single address returned by resolve() (still used for SSL engine setup). private static final AtomicInteger OFFSET = new AtomicInteger(); + // Rotates the starting candidate of resolveAll(). Kept separate from OFFSET so that SSL engine + // creation (which calls the deprecated resolve() once per connection) does not advance the + // resolveAll() counter a second time -- otherwise the start index would move in steps of 2 and + // rotation would collapse to a single IP whenever the proxy resolves to an even number of + // A-records (DRIVER-201). + private static final AtomicInteger RESOLVE_ALL_OFFSET = new AtomicInteger(); private final InetSocketAddress proxyAddress; private final String serverName; @@ -75,6 +86,57 @@ public InetSocketAddress resolve() { } } + /** + * Returns all socket addresses for this SNI proxy endpoint. + * + *

Re-resolves the proxy hostname on each call and returns one {@link InetSocketAddress} per + * A-record, so that the driver can try every proxy IP in sequence if one is unreachable. + * + *

All A-records are returned so a single connection attempt can fall back across every proxy + * IP, but the candidate order is rotated on each call using a dedicated round-robin counter + * ({@link #RESOLVE_ALL_OFFSET}). This spreads healthy connections across proxy IPs instead of + * always starting at the first one, preserving the previous load-balancing behavior. The counter + * is intentionally separate from the one used by {@link #resolve()} so that SSL engine setup + * (which calls {@code resolve()} once per connection) does not perturb this rotation. + * + *

The blocking DNS lookup is run on {@code executor} so it never blocks the calling (event + * loop) thread; the returned stage completes with the rotated candidate array, or completes + * exceptionally with an {@link IllegalArgumentException} if the proxy hostname cannot be + * resolved. + */ + @NonNull + @Override + public CompletionStage resolveAll(@NonNull Executor executor) { + return CompletableFuture.supplyAsync( + () -> { + try { + InetAddress[] aRecords = InetAddress.getAllByName(proxyAddress.getHostName()); + if (aRecords.length == 0) { + throw new IllegalArgumentException( + "Could not resolve proxy address " + proxyAddress.getHostName()); + } + // The order of the returned addresses is unspecified. Sort by IP so the round-robin + // rotation below is deterministic across calls. + Arrays.sort(aRecords, IP_COMPARATOR); + int start = + (aRecords.length == 1) + ? 0 + : RESOLVE_ALL_OFFSET.getAndUpdate(x -> x == Integer.MAX_VALUE ? 0 : x + 1) + % aRecords.length; + SocketAddress[] result = new SocketAddress[aRecords.length]; + for (int i = 0; i < aRecords.length; i++) { + InetAddress aRecord = aRecords[(start + i) % aRecords.length]; + result[i] = new InetSocketAddress(aRecord, proxyAddress.getPort()); + } + return result; + } catch (UnknownHostException e) { + throw new IllegalArgumentException( + "Could not resolve proxy address " + proxyAddress.getHostName(), e); + } + }, + executor); + } + @Override public boolean equals(Object other) { if (other == this) { 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/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java b/core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java index c9fee86f2c1..b2ed5111077 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/session/DefaultSession.java @@ -630,6 +630,14 @@ private void onChildrenClosed(List> childrenCloseStages) { for (CompletionStage stage : childrenCloseStages) { warnIfFailed(stage); } + // The channel factory owns a DNS-resolver executor that is not an AsyncAutoCloseable child; + // shut it down here, after all pools/control-connection that used it are closed. Guarded like + // the other context-component accesses below (the factory may have failed to initialize). + try { + context.getChannelFactory().close(); + } catch (Throwable t) { + // ignore: the factory may have failed to initialize, nothing to close + } context .getNettyOptions() .onClose() diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java index 343d3f9e4e7..c64a334b1fe 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/DefaultSslEngineFactory.java @@ -116,6 +116,8 @@ protected String hostNoLookup(InetSocketAddress addr) { @NonNull @Override + @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); SSL + // factories legitimately need a single address for hostname verification. public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) { SSLEngine engine; SocketAddress remoteAddress = remoteEndpoint.resolve(); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java index 4d2cb69fbfc..424120a99f3 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/ssl/SniSslEngineFactory.java @@ -51,6 +51,8 @@ public SniSslEngineFactory(SSLContext sslContext, boolean allowDnsReverseLookupS @NonNull @Override + @SuppressWarnings("deprecation") // resolve() is deprecated in favour of resolveAll(); SSL + // factories legitimately need a single address for SNI hostname verification. public SSLEngine newSslEngine(@NonNull EndPoint remoteEndpoint) { if (!(remoteEndpoint instanceof SniEndPoint)) { throw new IllegalArgumentException( diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf index 721afa1cd76..b154f888481 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1204,27 +1204,20 @@ datastax-java-driver { } - # Whether to resolve the addresses passed to `basic.contact-points`. + # DEPRECATED: this option no longer has any effect and will be removed in a future release. # - # If this is true, addresses are created with `InetSocketAddress(String, int)`: the host name will - # be resolved the first time, and the driver will use the resolved IP address for all subsequent - # connection attempts. + # Contact points are now always kept as unresolved hostnames and expanded to all of their + # DNS-mapped IPs lazily at connection time (see EndPoint.resolveAll()). This means the driver + # tries every IP a hostname resolves to, and re-resolves the hostname on each new connection so + # DNS changes are picked up automatically. Previously this option selected between resolving a + # contact-point hostname once (true) and re-resolving it on every connection (false); that + # distinction no longer applies. # - # If this is false, addresses are created with `InetSocketAddress.createUnresolved()`: the host - # name will be resolved again every time the driver opens a new connection. This is useful for - # containerized environments where DNS records are more likely to change over time (note that the - # JVM and OS have their own DNS caching mechanisms, so you might need additional configuration - # beyond the driver). + # This option only ever applied to the contact points specified in the configuration -- never to + # programmatic contact points passed to SessionBuilder.addContactPoints, nor to dynamically + # discovered peers. # - # This option only applies to the contact points specified in the configuration. It has no effect - # on: - # - programmatic contact points passed to SessionBuilder.addContactPoints: these addresses are - # built outside of the driver, so it is your responsibility to provide unresolved instances. - # - dynamically discovered peers: the driver relies on Cassandra system tables, which expose raw - # IP addresses. Use a custom address translator to convert them to unresolved addresses (if - # you're in a containerized environment, you probably already need address translation anyway). - # - # Required: no (defaults to false) + # Required: no # Modifiable at runtime: no # Overridable in a profile: no advanced.resolve-contact-points = false @@ -2318,14 +2311,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 connection time (see + # EndPoint.resolveAll()). 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/oss/driver/internal/core/channel/ChannelFactoryAsyncResolveTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryAsyncResolveTest.java new file mode 100644 index 00000000000..d604e4709fb --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryAsyncResolveTest.java @@ -0,0 +1,146 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.channel; + +import static com.datastax.oss.driver.Assertions.assertThat; +import static com.datastax.oss.driver.Assertions.assertThatStage; +import static org.mockito.Mockito.when; + +import com.datastax.oss.driver.api.core.DefaultProtocolVersion; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.metadata.EndPoint; +import com.datastax.oss.driver.internal.core.metrics.NoopNodeMetricUpdater; +import edu.umd.cs.findbugs.annotations.NonNull; +import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; +import org.junit.Test; + +/** + * Verifies that {@link ChannelFactory#connect} resolves endpoint addresses asynchronously, off the + * calling (admin event loop) thread, so DNS never blocks the caller. + */ +public class ChannelFactoryAsyncResolveTest extends ChannelFactoryTestBase { + + @Test + public void should_resolve_on_dedicated_resolver_thread() throws Exception { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + CompletableFuture resolverThreadName = new CompletableFuture<>(); + EndPoint endPoint = + new EndPoint() { + @NonNull + @Override + public SocketAddress resolve() { + return SERVER_ADDRESS.resolve(); + } + + @NonNull + @Override + public CompletionStage resolveAll(@NonNull Executor executor) { + return CompletableFuture.supplyAsync( + () -> { + resolverThreadName.complete(Thread.currentThread().getName()); + return new SocketAddress[] {SERVER_ADDRESS.resolve()}; + }, + executor); + } + + @NonNull + @Override + public String asMetricPrefix() { + return "test"; + } + }; + + String callerThreadName = Thread.currentThread().getName(); + + // When + CompletionStage channelFuture = + factory.connect( + endPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + completeSimpleChannelInit(); + + // Then – resolution ran on the factory's dedicated resolver executor, not the caller thread + String actualThreadName = resolverThreadName.get(2, TimeUnit.SECONDS); + assertThat(actualThreadName).isNotEqualTo(callerThreadName); + assertThat(actualThreadName).contains("-connection-resolver-"); + assertThatStage(channelFuture).isSuccess(); + } + + @Test + public void should_not_block_caller_while_resolving() throws Exception { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + CountDownLatch resolutionEntered = new CountDownLatch(1); + CountDownLatch releaseResolution = new CountDownLatch(1); + EndPoint endPoint = + new EndPoint() { + @NonNull + @Override + public SocketAddress resolve() { + return SERVER_ADDRESS.resolve(); + } + + @NonNull + @Override + public CompletionStage resolveAll(@NonNull Executor executor) { + return CompletableFuture.supplyAsync( + () -> { + resolutionEntered.countDown(); + try { + releaseResolution.await(5, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return new SocketAddress[] {SERVER_ADDRESS.resolve()}; + }, + executor); + } + + @NonNull + @Override + public String asMetricPrefix() { + return "test"; + } + }; + + // When + CompletionStage channelFuture = + factory.connect( + endPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + + // Then – connect() returned control while resolution is still blocked (caller not blocked) + assertThat(resolutionEntered.await(2, TimeUnit.SECONDS)).isTrue(); + assertThat(channelFuture.toCompletableFuture().isDone()).isFalse(); + + // Once resolution is unblocked the connection proceeds to completion. + releaseResolution.countDown(); + completeSimpleChannelInit(); + assertThatStage(channelFuture).isSuccess(); + } +} diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryMultiAddressTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryMultiAddressTest.java new file mode 100644 index 00000000000..5e56c7c07c9 --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryMultiAddressTest.java @@ -0,0 +1,117 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.channel; + +import static com.datastax.oss.driver.Assertions.assertThatStage; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.when; + +import com.datastax.oss.driver.api.core.DefaultProtocolVersion; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.metadata.EndPoint; +import com.datastax.oss.driver.internal.core.metrics.NoopNodeMetricUpdater; +import edu.umd.cs.findbugs.annotations.NonNull; +import io.netty.channel.local.LocalAddress; +import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; +import org.junit.Test; + +/** + * Verifies that {@link ChannelFactory#connect} tries every candidate returned by {@link + * EndPoint#resolveAll(java.util.concurrent.Executor)} in sequence: it falls back to the next + * address when one is unreachable, and only fails the overall future once all candidates are + * exhausted, carrying the earlier failures as suppressed exceptions. + */ +public class ChannelFactoryMultiAddressTest extends ChannelFactoryTestBase { + + // A local address that no server is bound to: connecting to it fails immediately. + private static final SocketAddress UNREACHABLE_1 = + new LocalAddress(ChannelFactoryMultiAddressTest.class.getSimpleName() + "-unreachable-1"); + private static final SocketAddress UNREACHABLE_2 = + new LocalAddress(ChannelFactoryMultiAddressTest.class.getSimpleName() + "-unreachable-2"); + + @Test + public void should_fall_back_to_next_candidate_when_first_is_unreachable() { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + // First candidate is unreachable, second is the running local server. + EndPoint endPoint = endPointReturning(UNREACHABLE_1, SERVER_ADDRESS.resolve()); + + // When + CompletionStage channelFuture = + factory.connect( + endPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + // The handshake only happens once we fall back to the reachable second candidate. + completeSimpleChannelInit(); + + // Then – the connection succeeds via the second candidate. + assertThatStage(channelFuture).isSuccess(); + } + + @Test + public void should_fail_with_suppressed_causes_when_all_candidates_are_unreachable() { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + EndPoint endPoint = endPointReturning(UNREACHABLE_1, UNREACHABLE_2); + + // When + CompletionStage channelFuture = + factory.connect( + endPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + + // Then – the future fails, and the earlier candidate's failure is preserved as a suppressed + // exception on the last candidate's error (rather than being silently dropped). + assertThatStage(channelFuture) + .isFailed( + e -> + assertThat(e.getSuppressed()) + .as("earlier candidate failures should be attached as suppressed exceptions") + .isNotEmpty()); + } + + /** An endpoint whose {@link EndPoint#resolveAll} yields the given candidates, in order. */ + private static EndPoint endPointReturning(SocketAddress... candidates) { + return new EndPoint() { + @NonNull + @Override + public SocketAddress resolve() { + return candidates[candidates.length - 1]; + } + + @NonNull + @Override + public CompletionStage resolveAll(@NonNull Executor executor) { + return CompletableFuture.completedFuture(candidates.clone()); + } + + @NonNull + @Override + public String asMetricPrefix() { + return "test"; + } + }; + } +} diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryResolveAllGuardTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryResolveAllGuardTest.java new file mode 100644 index 00000000000..99fbd156f7a --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryResolveAllGuardTest.java @@ -0,0 +1,136 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.channel; + +import static com.datastax.oss.driver.Assertions.assertThatStage; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.datastax.oss.driver.api.core.DefaultProtocolVersion; +import com.datastax.oss.driver.api.core.config.DefaultDriverOption; +import com.datastax.oss.driver.api.core.metadata.EndPoint; +import com.datastax.oss.driver.internal.core.metrics.NoopNodeMetricUpdater; +import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionException; +import java.util.concurrent.CompletionStage; +import org.junit.Test; + +/** + * Verifies that {@link ChannelFactory#connect} completes the result future exceptionally (rather + * than throwing or hanging) when {@link EndPoint#resolveAll(java.util.concurrent.Executor)} + * completes with {@code null}, an empty array, throws synchronously, or completes exceptionally. + * Resolution is now asynchronous, so these assertions exercise the {@code whenComplete} callback. + */ +public class ChannelFactoryResolveAllGuardTest extends ChannelFactoryTestBase { + + @Test + public void should_fail_future_when_resolve_all_returns_null() { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + EndPoint badEndPoint = mock(EndPoint.class); + when(badEndPoint.resolveAll(any())).thenReturn(CompletableFuture.completedFuture(null)); + + // When + CompletionStage channelFuture = + factory.connect( + badEndPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + + // Then – future must complete exceptionally without hanging + assertThatStage(channelFuture) + .isFailed( + e -> + assertThat(e) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("resolveAll() must return a non-null, non-empty array")); + } + + @Test + public void should_fail_future_when_resolve_all_returns_empty_array() { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + EndPoint badEndPoint = mock(EndPoint.class); + when(badEndPoint.resolveAll(any())) + .thenReturn(CompletableFuture.completedFuture(new SocketAddress[0])); + + // When + CompletionStage channelFuture = + factory.connect( + badEndPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + + // Then – future must complete exceptionally without hanging + assertThatStage(channelFuture) + .isFailed( + e -> + assertThat(e) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("resolveAll() must return a non-null, non-empty array")); + } + + @Test + public void should_fail_future_when_resolve_all_throws_synchronously() { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + EndPoint badEndPoint = mock(EndPoint.class); + RuntimeException testException = new RuntimeException("DNS lookup failed"); + when(badEndPoint.resolveAll(any())).thenThrow(testException); + + // When + CompletionStage channelFuture = + factory.connect( + badEndPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + + // Then – future must complete exceptionally with the thrown exception + assertThatStage(channelFuture).isFailed(e -> assertThat(e).isSameAs(testException)); + } + + @Test + public void should_fail_future_with_unwrapped_cause_when_resolve_all_stage_fails() { + // Given + when(defaultProfile.isDefined(DefaultDriverOption.PROTOCOL_VERSION)).thenReturn(false); + when(protocolVersionRegistry.highestNonBeta()).thenReturn(DefaultProtocolVersion.V4); + ChannelFactory factory = newChannelFactory(); + + EndPoint badEndPoint = mock(EndPoint.class); + RuntimeException testException = new RuntimeException("DNS lookup failed"); + // A real impl offloads via supplyAsync, which wraps the supplier's throwable in a + // CompletionException; connect() must unwrap it so callers see the original cause. + CompletableFuture failedStage = new CompletableFuture<>(); + failedStage.completeExceptionally(new CompletionException(testException)); + when(badEndPoint.resolveAll(any())).thenReturn(failedStage); + + // When + CompletionStage channelFuture = + factory.connect( + badEndPoint, null, null, DriverChannelOptions.DEFAULT, NoopNodeMetricUpdater.INSTANCE); + + // Then – future must complete exceptionally with the unwrapped original exception + assertThatStage(channelFuture).isFailed(e -> assertThat(e).isSameAs(testException)); + } +} diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/LocalEndPoint.java b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/LocalEndPoint.java index c90731eece9..e9c8f836de4 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/channel/LocalEndPoint.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/channel/LocalEndPoint.java @@ -21,6 +21,9 @@ import edu.umd.cs.findbugs.annotations.NonNull; import io.netty.channel.local.LocalAddress; import java.net.SocketAddress; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.Executor; /** Endpoint implementation for unit tests that use the local Netty transport. */ public class LocalEndPoint implements EndPoint { @@ -37,6 +40,13 @@ public SocketAddress resolve() { return localAddress; } + @NonNull + @Override + public CompletionStage resolveAll(@NonNull Executor executor) { + // In-memory local address; resolve synchronously without hopping to the executor. + return CompletableFuture.completedFuture(new SocketAddress[] {localAddress}); + } + @NonNull @Override public String asMetricPrefix() { 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/DefaultEndPointTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPointTest.java index 7da8fb39415..404de8f7f4c 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPointTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPointTest.java @@ -20,7 +20,12 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.UnknownHostException; +import java.util.HashSet; +import java.util.Set; import org.junit.Test; public class DefaultEndPointTest { @@ -57,4 +62,53 @@ public void should_reject_null_address() { .isInstanceOf(NullPointerException.class) .hasMessage("address can't be null"); } + + @Test + public void resolve_all_returns_single_element_for_already_resolved_address() { + DefaultEndPoint endPoint = new DefaultEndPoint(new InetSocketAddress("127.0.0.1", 9042)); + SocketAddress[] all = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + assertThat(all).hasSize(1); + assertThat(((InetSocketAddress) all[0]).isUnresolved()).isFalse(); + assertThat(((InetSocketAddress) all[0]).getHostString()).isEqualTo("127.0.0.1"); + } + + @Test + public void resolve_all_expands_unresolved_hostname_to_all_dns_ips() throws UnknownHostException { + // localhost reliably resolves to at least 127.0.0.1 (and possibly ::1). + DefaultEndPoint endPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("localhost", 9042)); + + SocketAddress[] all = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + + // The complete DNS result set must be expanded -- one resolved InetSocketAddress per record -- + // so a connection attempt can fall back across every IP, not just the first. Compare against an + // independent getAllByName() lookup so the test fails if production returned only a subset. + InetAddress[] expectedIps = InetAddress.getAllByName("localhost"); + Set expected = new HashSet<>(); + for (InetAddress ip : expectedIps) { + expected.add(new InetSocketAddress(ip, 9042)); + } + Set actual = new HashSet<>(); + for (SocketAddress addr : all) { + InetSocketAddress inet = (InetSocketAddress) addr; + assertThat(inet.isUnresolved()).isFalse(); + assertThat(inet.getPort()).isEqualTo(9042); + actual.add(inet); + } + assertThat(all).hasSize(expectedIps.length); + assertThat(actual).isEqualTo(expected); + } + + @Test + public void resolve_all_falls_back_to_single_element_when_hostname_is_unresolvable() { + // Unresolvable hostname: resolveAll() must not throw; it returns the unresolved address. + DefaultEndPoint endPoint = + new DefaultEndPoint( + InetSocketAddress.createUnresolved("this-host-does-not-exist.invalid", 9042)); + SocketAddress[] all = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + assertThat(all).hasSize(1); + // The fallback address is the original unresolved one. + assertThat(((InetSocketAddress) all[0]).getHostString()) + .isEqualTo("this-host-does-not-exist.invalid"); + } } 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..8c635982eea 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 @@ -40,10 +40,11 @@ import com.datastax.oss.driver.internal.core.context.EventBus; import com.datastax.oss.driver.internal.core.context.InternalDriverContext; import com.datastax.oss.driver.internal.core.metrics.MetricsFactory; +import com.datastax.oss.driver.internal.core.util.collection.QueryPlan; +import com.datastax.oss.driver.internal.core.util.collection.SimpleQueryPlan; import com.datastax.oss.driver.shaded.guava.common.collect.ImmutableList; 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.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; @@ -102,13 +104,16 @@ public void setup() { when(metadata.getNodes()).thenReturn(allNodes); when(metadataManager.getContactPoints()).thenReturn(contactPoints); when(context.getMetadataManager()).thenReturn(metadataManager); + when(context.getTopologyMonitor()).thenReturn(topologyMonitor); when(context.getConfig()).thenReturn(config); when(config.getDefaultProfile()).thenReturn(defaultProfile); when(defaultProfile.getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS)) .thenReturn(false); - defaultPolicyQueryPlan = Lists.newLinkedList(ImmutableList.of(node3, node2, node1)); + // Use a real built-in QueryPlan (not a mutable LinkedList): its add()/addAll() throw, so the + // control-reconnection plan must compose rather than mutate it (see CompositeQueryPlan usage). + defaultPolicyQueryPlan = new SimpleQueryPlan(node3, node2, node1); when(policy1.newQueryPlan(null, null)).thenReturn(defaultPolicyQueryPlan); eventBus = spy(new EventBus("test")); @@ -130,26 +135,28 @@ public void setup() { @Test public void should_build_control_connection_query_plan_from_contact_points_before_init() { - // When + // When — before init, the control-reconnection plan is built straight from the contact points + // (bypassing the load balancing policies), so each hostname can be tried on the first connect. Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); - // Then + // Then — query plan contains the contact points, and no policy was consulted for (LoadBalancingPolicy policy : ImmutableList.of(policy1, policy2, policy3)) { verify(policy, never()).newQueryPlan(null, null); } - assertThat(queryPlan).hasSameElementsAs(contactPoints); + assertThat(queryPlan).containsExactlyInAnyOrder(node1, node2); } @Test public void should_build_query_plan_from_contact_points_before_init() { - // When + // When — before init, the query plan is built straight from the contact points (bypassing the + // load balancing policies) Queue queryPlan = wrapper.newQueryPlan(null, DriverExecutionProfile.DEFAULT_NAME, null); - // Then + // Then — query plan contains the contact points, and no policy was consulted for (LoadBalancingPolicy policy : ImmutableList.of(policy1, policy2, policy3)) { verify(policy, never()).newQueryPlan(null, null); } - assertThat(queryPlan).hasSameElementsAs(contactPoints); + assertThat(queryPlan).containsExactlyInAnyOrder(node1, node2); } @Test @@ -204,8 +211,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 original contact points appended at the end. Set remainingEndpoints = new java.util.HashSet<>(); for (Node n : queryPlan) { remainingEndpoints.add(n.getEndPoint()); @@ -217,14 +223,92 @@ public void should_fetch_control_connection_query_plan_from_policy_after_init() assertThat(remainingEndpoints).isEqualTo(contactEndpoints); } + @Test + public void should_not_duplicate_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 the 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. + + // When + Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); + + // Then — the contact points are read 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)).getContactPoints(); + 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_append_contact_points_when_query_plan_empty_even_if_topology_monitor_reresolves() { + // Given — the flag is enabled and the topology monitor re-resolves node addresses on its own, + // but the live-node query plan is empty. With no node to try, reconnection can only recover + // through the contact-point fallback, so it must be appended despite the re-resolving monitor. + when(defaultProfile.getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS)) + .thenReturn(true); + when(topologyMonitor.reresolvesNodeAddresses()).thenReturn(true); + wrapper.init(); + when(policy1.newQueryPlan(null, null)).thenReturn(QueryPlan.EMPTY); + + // When + Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); + + // Then — the contact points are appended (compare by endpoint since they are new instances) + assertThat(queryPlan.size()).isEqualTo(contactPoints.size()); + Set resultEndpoints = new java.util.HashSet<>(); + for (Node n : queryPlan) { + resultEndpoints.add(n.getEndPoint()); + } + Set contactEndpoints = new java.util.HashSet<>(); + for (DefaultNode n : contactPoints) { + contactEndpoints.add(n.getEndPoint()); + } + assertThat(resultEndpoints).isEqualTo(contactEndpoints); + } + @Test public void should_return_contact_points_when_query_plan_empty_and_flag_enabled() { // Given when(defaultProfile.getBoolean(DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS)) .thenReturn(true); wrapper.init(); - // Make the policy return an empty query plan - when(policy1.newQueryPlan(null, null)).thenReturn(Lists.newLinkedList(ImmutableList.of())); + // Make the policy return an empty query plan (QueryPlan.EMPTY, as the real policies do) + when(policy1.newQueryPlan(null, null)).thenReturn(QueryPlan.EMPTY); // When Queue queryPlan = wrapper.newControlReconnectionQueryPlan(); diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/SniEndPointTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/SniEndPointTest.java new file mode 100644 index 00000000000..99672006747 --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/SniEndPointTest.java @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datastax.oss.driver.internal.core.metadata; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Set; +import org.junit.Test; + +public class SniEndPointTest { + + @Test + public void resolve_all_returns_all_proxy_addresses_for_resolvable_hostname() { + // localhost reliably resolves to at least one address + SniEndPoint endPoint = + new SniEndPoint(new InetSocketAddress("localhost", 9042), "test-server-name"); + SocketAddress[] all = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + assertThat(all).isNotEmpty(); + for (SocketAddress addr : all) { + InetSocketAddress inet = (InetSocketAddress) addr; + assertThat(inet.isUnresolved()).isFalse(); + assertThat(inet.getPort()).isEqualTo(9042); + } + } + + @Test + public void resolve_all_fails_for_unresolvable_hostname() { + SniEndPoint endPoint = + new SniEndPoint( + new InetSocketAddress("this-host-does-not-exist.invalid", 9042), "test-server-name"); + // Resolution is async now: the failure surfaces as an exceptionally-completed stage whose cause + // is the IllegalArgumentException. + assertThatThrownBy(() -> endPoint.resolveAll(Runnable::run).toCompletableFuture().join()) + .hasCauseInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Could not resolve proxy address"); + } + + @Test + public void resolve_returns_single_address_from_round_robin() { + // Sanity check: resolve() still works and returns a single address + SniEndPoint endPoint = + new SniEndPoint(new InetSocketAddress("localhost", 9042), "test-server-name"); + InetSocketAddress addr = endPoint.resolve(); + assertThat(addr.isUnresolved()).isFalse(); + assertThat(addr.getPort()).isEqualTo(9042); + } + + @Test + public void resolve_all_returns_complete_and_rotated_candidate_order() { + // resolveAll() must always return the full set of A-records (so a single connection attempt can + // fall back across every proxy IP), while rotating the starting element on each call to + // preserve the round-robin behavior of resolve(). We assert both invariants without depending + // on how many addresses "localhost" resolves to in a given environment. + SniEndPoint endPoint = + new SniEndPoint(new InetSocketAddress("localhost", 9042), "test-server-name"); + + SocketAddress[] first = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + int size = first.length; + assertThat(size).isGreaterThanOrEqualTo(1); + Set expected = new HashSet<>(Arrays.asList(first)); + + // Every call returns the same complete set of candidates, regardless of rotation. + boolean sawRotation = false; + SocketAddress firstElementSeed = first[0]; + for (int call = 0; call < size * 2; call++) { + SocketAddress[] candidates = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + assertThat(candidates).hasSize(size); + assertThat(new HashSet<>(Arrays.asList(candidates))).isEqualTo(expected); + if (!candidates[0].equals(firstElementSeed)) { + sawRotation = true; + } + } + + // When there is more than one A-record the starting candidate must rotate across calls. + if (size > 1) { + assertThat(sawRotation) + .as("resolveAll() should rotate the starting candidate when multiple IPs exist") + .isTrue(); + } + } + + @Test + public void resolve_all_rotation_is_not_disturbed_by_interleaved_resolve() { + // resolveAll() rotates using a counter independent from resolve(). SSL engine setup calls the + // deprecated resolve() once per connection; if the two shared a counter, that extra advance + // would pin resolveAll()'s start index to 0 whenever the proxy resolves to an even number of + // A-records (DRIVER-201). Interleaving resolve() calls here must not stop resolveAll() from + // rotating. On single-address environments (size == 1) rotation is a no-op, matching the + // conditional assertion in resolve_all_returns_complete_and_rotated_candidate_order(). + SniEndPoint endPoint = + new SniEndPoint(new InetSocketAddress("localhost", 9042), "test-server-name"); + + SocketAddress[] first = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + int size = first.length; + SocketAddress firstStart = first[0]; + + boolean sawRotation = false; + for (int call = 0; call < size * 2; call++) { + SocketAddress[] candidates = endPoint.resolveAll(Runnable::run).toCompletableFuture().join(); + if (!candidates[0].equals(firstStart)) { + sawRotation = true; + } + // Interleave a resolve() call, as the SSL engine does on every connection. + endPoint.resolve(); + } + + if (size > 1) { + assertThat(sawRotation) + .as("resolveAll() rotation must survive interleaved resolve() calls") + .isTrue(); + } + } +} 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..730827b2537 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 @@ -114,6 +114,45 @@ public void should_connect_with_mocked_hostname() { } } + @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); + } + } + } + @Test public void replace_cluster_test() { final int numberOfNodes = 3; diff --git a/upgrade_guide/README.md b/upgrade_guide/README.md index 214399dacc7..f1dfdceb82e 100644 --- a/upgrade_guide/README.md +++ b/upgrade_guide/README.md @@ -19,6 +19,27 @@ under the License. ## Upgrade guide +### 4.19.2.1 + +#### Contact points are expanded to all their DNS addresses at connection time + +Contact points backed by a hostname are now kept unresolved and expanded to **all** the IP +addresses the hostname maps to, at connection time (via `EndPoint.resolveAll()`). Previously only +the first address returned by DNS was tried, so a single non-responsive IP behind a multi-record +hostname could fail the initial connection (or a control-connection reconnect) even when the other +addresses were healthy. No configuration change is required to benefit from this. + +As part of this change: + +- `advanced.resolve-contact-points` is deprecated and now has **no effect**. Contact points are + always kept as unresolved hostnames and expanded at connection time. An already-resolved + `InetSocketAddress` passed programmatically is still used as provided, with no further expansion. +- `advanced.control-connection.reconnection.fallback-to-original-contact-points` now defaults to + `true` (previously `false`). This is also the driver's DNS re-resolution path: metadata nodes + hold an already-resolved endpoint that is never re-resolved, so on control-connection reconnect + the driver falls back to the original contact points to pick up current DNS records once the + live-node plan is exhausted. Set it to `false` to restore the previous behavior. + ### 4.19.0.7 #### Cloud private-endpoint support via client routes