From 64fcb750fcbfbc9be8314ad6d53e69a535e39665 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 23:23:36 +0200 Subject: [PATCH 1/9] refactor: remove dead local-DC contact-point compatibility check (DRIVER-201) OptionalLocalDcHelper.checkLocalDatacenterCompatibility() warned when a contact point reported a datacenter different from the configured local DC. This has been dead code on scylla-4.x since 12e6acb90b: refresh matches nodes by hostId only, so contact-point nodes never get a datacenter assigned and the warning could never fire. Remove it. The separate "configured local DC matches no node" warning is retained. Co-Authored-By: Claude Opus 4.8 --- .../helper/OptionalLocalDcHelper.java | 74 +++++-------------- 1 file changed, 17 insertions(+), 57 deletions(-) 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); } /** From 2a5abe199350e6716d58a12debc68771ab835b89 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 23:23:59 +0200 Subject: [PATCH 2/9] feat: keep contact points unresolved; deprecate RESOLVE_CONTACT_POINTS (DRIVER-201) Contact points backed by a hostname are now always kept unresolved, so they can be expanded to all their DNS-mapped IPs later, at connection time (see the follow-up EndPoint.resolveAll() commits). SessionBuilder no longer reads RESOLVE_CONTACT_POINTS when merging contact points; the option is deprecated and has no effect. An already-resolved InetSocketAddress passed programmatically is still used as provided, with no further expansion. Co-Authored-By: Claude Opus 4.8 --- .../api/core/config/DefaultDriverOption.java | 4 ++++ .../api/core/config/TypedDriverOption.java | 8 ++++++- .../api/core/session/SessionBuilder.java | 21 +++++++++++-------- core/src/main/resources/reference.conf | 4 ++++ 4 files changed, 27 insertions(+), 10 deletions(-) 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..6efcbcb35ca 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 @@ -837,7 +837,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/TypedDriverOption.java b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java index e7c606cade8..ccc9c5d81b2 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 @@ -664,7 +664,13 @@ public String toString() { /** The coalescer reschedule interval. */ public static final TypedDriverOption COALESCER_INTERVAL = new TypedDriverOption<>(DefaultDriverOption.COALESCER_INTERVAL, GenericType.DURATION); - /** Whether to resolve the addresses passed to `basic.contact-points`. */ + /** + * Whether to resolve the addresses passed to `basic.contact-points`. + * + * @deprecated Contact points are now always kept as unresolved hostnames and expanded to all + * their DNS-mapped IPs lazily at connection time. Setting this option has no effect. + */ + @Deprecated public static final TypedDriverOption RESOLVE_CONTACT_POINTS = new TypedDriverOption<>(DefaultDriverOption.RESOLVE_CONTACT_POINTS, GenericType.BOOLEAN); /** diff --git a/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java b/core/src/main/java/com/datastax/oss/driver/api/core/session/SessionBuilder.java index 8375f0ef30b..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/resources/reference.conf b/core/src/main/resources/reference.conf index 721afa1cd76..a29f5d5e908 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1227,6 +1227,10 @@ datastax-java-driver { # Required: no (defaults to false) # Modifiable at runtime: no # Overridable in a profile: no + # + # DEPRECATED: this option no longer has any effect. Contact points are always kept as unresolved + # hostnames and expanded to all their DNS-mapped IPs at connection time (see + # EndPoint.resolveAll()). It will be removed in a future release. advanced.resolve-contact-points = false advanced.protocol { From b67f9c00cc7548ce40d6c51ccb80b0bf66061a32 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 23:24:33 +0200 Subject: [PATCH 3/9] feat: add EndPoint.resolveAll() for multi-address DNS expansion (DRIVER-201) Add EndPoint.resolveAll(Executor), a default method returning a CompletionStage of all socket addresses an endpoint maps to, so a hostname with multiple A-records can be expanded to every IP. The default offloads resolve() to the executor and wraps it in a single-element array, keeping existing third-party implementations working (source- and binary-compatible; resolve() is not deprecated yet). Overrides: - DefaultEndPoint: InetAddress.getAllByName() expansion for unresolved addresses, falling back to the single stored address on DNS failure. - SniEndPoint: returns the complete A-record set, rotated with the same round-robin OFFSET as resolve() so healthy connections spread across proxy IPs. - ClientRoutesEndPoint: single-address-by-design, delegates to resolve(). Co-Authored-By: Claude Opus 4.8 --- .../driver/api/core/metadata/EndPoint.java | 45 +++++++- .../core/metadata/ClientRoutesEndPoint.java | 18 ++++ .../core/metadata/DefaultEndPoint.java | 50 +++++++++ .../internal/core/metadata/SniEndPoint.java | 53 +++++++++ .../core/metadata/DefaultEndPointTest.java | 37 +++++++ .../core/metadata/SniEndPointTest.java | 101 ++++++++++++++++++ 6 files changed, 302 insertions(+), 2 deletions(-) create mode 100644 core/src/test/java/com/datastax/oss/driver/internal/core/metadata/SniEndPointTest.java 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..c4b131d9e6f 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,14 +18,16 @@ 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. */ @@ -40,6 +42,45 @@ public interface EndPoint { @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/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/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/SniEndPoint.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/SniEndPoint.java index d1ab8eec98d..acd3431580e 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,10 +22,14 @@ 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 { @@ -75,6 +79,55 @@ 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 the same round-robin {@link #OFFSET} + * counter as {@link #resolve()}. This spreads healthy connections across proxy IPs instead of + * always starting at the first one, preserving the previous load-balancing behavior. + * + *

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 + : 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/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..66b794cdcf8 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 @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.net.InetSocketAddress; +import java.net.SocketAddress; import org.junit.Test; public class DefaultEndPointTest { @@ -57,4 +58,40 @@ 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_at_least_one_address() { + // localhost reliably resolves to at least 127.0.0.1 + DefaultEndPoint endPoint = + new DefaultEndPoint(InetSocketAddress.createUnresolved("localhost", 9042)); + 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_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/SniEndPointTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/SniEndPointTest.java new file mode 100644 index 00000000000..c7f5bb695e8 --- /dev/null +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/SniEndPointTest.java @@ -0,0 +1,101 @@ +/* + * 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(); + } + } +} From e93ef24c2c7511ab01f4928d3ebcbe6566bc40bc Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 23:25:04 +0200 Subject: [PATCH 4/9] feat: resolve endpoints off the event loop with multi-IP fallback; deprecate resolve() (DRIVER-201) ChannelFactory.connect() now resolves via EndPoint.resolveAll() on a dedicated daemon resolver executor, so blocking DNS never runs on the admin event loop. The returned candidates are tried serially (tryNextCandidate): each address is attempted via connectToAddress(), and on a per-address failure the next candidate is tried; only when all are exhausted does the overall future fail. Protocol-version negotiation (downgrade retries) stays scoped to a single address. A null/empty resolveAll() result fails the future rather than NPE/AIOOBE. DefaultSession shuts the resolver executor down on close. resolve() is now @Deprecated in favour of resolveAll(). The five internal callers that legitimately need a single canonical address (DefaultTopologyMonitor, InsightsClient, DseGssApiAuthProviderBase, DefaultSslEngineFactory, SniSslEngineFactory) are annotated @SuppressWarnings("deprecation") so the -Werror build stays green. Co-Authored-By: Claude Opus 4.8 --- .../core/auth/DseGssApiAuthProviderBase.java | 2 + .../core/insights/InsightsClient.java | 6 + .../driver/api/core/metadata/EndPoint.java | 8 +- .../internal/core/channel/ChannelFactory.java | 171 +++++++++++++++++- .../core/metadata/DefaultTopologyMonitor.java | 4 + .../internal/core/session/DefaultSession.java | 8 + .../core/ssl/DefaultSslEngineFactory.java | 2 + .../core/ssl/SniSslEngineFactory.java | 2 + .../ChannelFactoryAsyncResolveTest.java | 146 +++++++++++++++ .../ChannelFactoryResolveAllGuardTest.java | 136 ++++++++++++++ .../internal/core/channel/LocalEndPoint.java | 10 + .../driver/core/resolver/MockResolverIT.java | 39 ++++ 12 files changed, 525 insertions(+), 9 deletions(-) create mode 100644 core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryAsyncResolveTest.java create mode 100644 core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryResolveAllGuardTest.java 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/metadata/EndPoint.java b/core/src/main/java/com/datastax/oss/driver/api/core/metadata/EndPoint.java index c4b131d9e6f..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 @@ -34,11 +34,17 @@ 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(); 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..d2e05ebef85 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 @@ -62,8 +62,11 @@ 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 +110,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 +133,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 +175,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 +248,137 @@ 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); + }); + } + + /** + * 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. + */ + 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) { + + 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()); + tryNextCandidate( + endPoint, + shardingInfo, + shardId, + options, + nodeMetricUpdater, + currentVersion, + isNegotiating, + attemptedVersions, + resultFuture, + candidates, + index + 1); + } else { + // 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 +387,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 +447,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 +460,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 +470,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/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/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/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/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/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; From 81ef5635adb223c40b7ad10e1e8cdd62d4a52003 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 23:25:32 +0200 Subject: [PATCH 5/9] fix: gate contact-point reconnection fallback by topology monitor and enable it by default (DRIVER-201) fallback-to-original-contact-points now defaults to true. On a control-connection reconnect the original (unresolved) contact points are appended after the live-node plan; EndPoint.resolveAll() re-expands each hostname to its current DNS IPs, which is the driver's DNS re-resolution path (metadata nodes hold an already-resolved endpoint that is never re-resolved). The append is gated: - only in the RUNNING state (before that newQueryPlan already builds the plan from the contact points, so appending would duplicate them); - skipped when the topology monitor re-resolves node addresses itself (TopologyMonitor.reresolvesNodeAddresses(), overridden true by CloudTopologyMonitor and route-aware in ClientRoutesTopologyMonitor), unless the regular plan is empty, to avoid resurrecting nodes those proxy-based monitors authoritatively removed. The plan is now composed (CompositeQueryPlan + SimpleQueryPlan) rather than mutated, since a RUNNING-state built-in query plan rejects add()/addAll(). HeartbeatIT pins the option to false so the extra OPTIONS message does not skew heartbeat counts. Co-Authored-By: Claude Opus 4.8 --- .../api/core/config/DefaultDriverOption.java | 9 +- .../driver/api/core/config/OptionsMap.java | 2 +- .../api/core/config/TypedDriverOption.java | 2 +- .../metadata/ClientRoutesTopologyMonitor.java | 17 +++ .../core/metadata/CloudTopologyMonitor.java | 8 ++ .../metadata/LoadBalancingPolicyWrapper.java | 44 +++++-- .../core/metadata/MetadataManager.java | 7 ++ .../core/metadata/TopologyMonitor.java | 20 ++++ core/src/main/resources/reference.conf | 14 ++- .../ClientRoutesTopologyMonitorTest.java | 55 +++++++++ .../LoadBalancingPolicyWrapperTest.java | 108 ++++++++++++++++-- .../driver/core/heartbeat/HeartbeatIT.java | 4 + 12 files changed, 261 insertions(+), 29 deletions(-) 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 6efcbcb35ca..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 */ 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 ccc9c5d81b2..fa12f40fd3e 100644 --- a/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java +++ b/core/src/main/java/com/datastax/oss/driver/api/core/config/TypedDriverOption.java @@ -600,7 +600,7 @@ public String toString() { public static final TypedDriverOption CONTROL_CONNECTION_AGREEMENT_WARN = new TypedDriverOption<>( DefaultDriverOption.CONTROL_CONNECTION_AGREEMENT_WARN, GenericType.BOOLEAN); - /** Whether to forcibly try original contacts if no live nodes are available */ + /** Whether to append the original contact points to the reconnection plan (defaults to true) */ public static final TypedDriverOption CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS = new TypedDriverOption<>( DefaultDriverOption.CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS, GenericType.BOOLEAN); diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java index 1ffc35fd9f4..a93c34463bd 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitor.java @@ -21,6 +21,7 @@ import com.datastax.oss.driver.api.core.config.ClientRoutesConfig; import com.datastax.oss.driver.api.core.config.DefaultDriverOption; import com.datastax.oss.driver.api.core.metadata.EndPoint; +import com.datastax.oss.driver.api.core.metadata.Node; import com.datastax.oss.driver.internal.core.adminrequest.AdminRequestHandler; import com.datastax.oss.driver.internal.core.adminrequest.AdminResult; import com.datastax.oss.driver.internal.core.adminrequest.AdminRow; @@ -480,6 +481,22 @@ protected EndPoint buildNodeEndPoint( return new ClientRoutesEndPoint(this, hostId, broadcastInetAddress, fallback); } + @Override + public boolean reresolvesNodeAddresses() { + // ClientRoutesEndPoint re-resolves via the client route hostname on every connection attempt, + // but only when a route exists for that host_id (see ClientRoutesEndPoint#resolve()) -- for + // mixed/incomplete route sets it falls back to a static, non-re-resolving endpoint instead. + // Only report true when every currently-known node actually has a live route; otherwise the + // contact-point reconnection fallback must stay available for the nodes stuck on the fallback. + Map routes = resolvedRoutesCache.get(); + for (Node node : context.getMetadataManager().getMetadata().getNodes().values()) { + if (!routes.containsKey(node.getHostId())) { + return false; + } + } + return true; + } + /** * Builds the CQL query to fetch client routes. * diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java index 021824a9b16..013027e139d 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/CloudTopologyMonitor.java @@ -44,4 +44,12 @@ protected EndPoint buildNodeEndPoint( UUID hostId = Objects.requireNonNull(row.getUuid("host_id")); return new SniEndPoint(cloudProxyAddress, hostId.toString()); } + + @Override + public boolean reresolvesNodeAddresses() { + // Nodes are reached through the cloud SNI proxy via SniEndPoint, which re-resolves the proxy + // hostname on every connection attempt. Appending the original contact points as a DNS + // re-resolution fallback is therefore unnecessary and could resurrect removed nodes. + return true; + } } diff --git a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapper.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapper.java index f3f3e4fe346..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/TopologyMonitor.java b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/TopologyMonitor.java index 1bb8e343d96..420cd8e4f92 100644 --- a/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/TopologyMonitor.java +++ b/core/src/main/java/com/datastax/oss/driver/internal/core/metadata/TopologyMonitor.java @@ -141,4 +141,24 @@ public interface TopologyMonitor extends AsyncAutoCloseable { * {@link DefaultTopologyMonitor}) should override this method. */ default void resetColumnCaches() {} + + /** + * Whether this monitor re-resolves node addresses dynamically on every connection attempt (for + * example by re-resolving a proxy hostname each time), rather than relying on an endpoint address + * captured once at node-registration time. + * + *

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

The default implementation returns {@code false}, which is correct for {@link + * DefaultTopologyMonitor}, whose {@code DefaultEndPoint}s cache their resolved address and never + * re-resolve. Proxy-based monitors that re-resolve per call should override this to return {@code + * true}. + */ + default boolean reresolvesNodeAddresses() { + return false; + } } diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf index a29f5d5e908..24506b49bcc 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -2322,14 +2322,20 @@ datastax-java-driver { } reconnection { - # Whether to forcibly add original contact points held by MetadataManager to the reconnection plan, - # in case there is no live nodes available according to LBP. - # Experimental. + # Whether to append the original contact points held by MetadataManager to the reconnection + # plan, after the live nodes reported by the load balancing policy. + # + # This is also the driver's DNS re-resolution path. Contact points are kept as unresolved + # hostnames and expanded to their current DNS IPs at 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/metadata/ClientRoutesTopologyMonitorTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitorTest.java index a1ba4617ef5..0a6268f3378 100644 --- a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitorTest.java +++ b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/ClientRoutesTopologyMonitorTest.java @@ -28,6 +28,8 @@ import com.datastax.oss.driver.api.core.config.DriverConfig; import com.datastax.oss.driver.api.core.config.DriverExecutionProfile; import com.datastax.oss.driver.api.core.metadata.EndPoint; +import com.datastax.oss.driver.api.core.metadata.Metadata; +import com.datastax.oss.driver.api.core.metadata.Node; import com.datastax.oss.driver.internal.core.adminrequest.AdminResult; import com.datastax.oss.driver.internal.core.adminrequest.AdminRow; import com.datastax.oss.driver.internal.core.channel.DriverChannel; @@ -66,6 +68,8 @@ public class ClientRoutesTopologyMonitorTest { @Mock private ControlConnection controlConnection; @Mock private DriverConfig driverConfig; @Mock private DriverExecutionProfile defaultProfile; + @Mock private MetadataManager metadataManager; + @Mock private Metadata metadata; private TestableClientRoutesTopologyMonitor handler; @@ -220,6 +224,57 @@ public void should_refresh_updates_routes() throws UnknownHostException { assertThat(handler.resolve(hostId2)).isNotNull(); } + // ---- reresolvesNodeAddresses() ------------------------------------------- + + @Test + public void should_reresolve_when_all_known_nodes_have_client_routes() { + UUID hostId1 = UUID.randomUUID(); + UUID hostId2 = UUID.randomUUID(); + Node node1 = Mockito.mock(Node.class); + when(node1.getHostId()).thenReturn(hostId1); + Node node2 = Mockito.mock(Node.class); + when(node2.getHostId()).thenReturn(hostId2); + + when(context.getMetadataManager()).thenReturn(metadataManager); + when(metadataManager.getMetadata()).thenReturn(metadata); + when(metadata.getNodes()).thenReturn(ImmutableMap.of(hostId1, node1, hostId2, node2)); + + handler.setRoutes( + ImmutableMap.of( + hostId1, new ClientRouteRecord(hostId1, "127.0.0.1", 9042), + hostId2, new ClientRouteRecord(hostId2, "127.0.0.2", 9042))); + + assertThat(handler.reresolvesNodeAddresses()).isTrue(); + } + + @Test + public void should_not_reresolve_when_a_known_node_has_no_client_route() { + UUID hostId1 = UUID.randomUUID(); + UUID hostId2 = UUID.randomUUID(); + Node node1 = Mockito.mock(Node.class); + when(node1.getHostId()).thenReturn(hostId1); + Node node2 = Mockito.mock(Node.class); + when(node2.getHostId()).thenReturn(hostId2); + + when(context.getMetadataManager()).thenReturn(metadataManager); + when(metadataManager.getMetadata()).thenReturn(metadata); + when(metadata.getNodes()).thenReturn(ImmutableMap.of(hostId1, node1, hostId2, node2)); + + // Only node1 has a live client route; node2 would fall back to a static endpoint. + handler.setRoutes(ImmutableMap.of(hostId1, new ClientRouteRecord(hostId1, "127.0.0.1", 9042))); + + assertThat(handler.reresolvesNodeAddresses()).isFalse(); + } + + @Test + public void should_reresolve_when_no_nodes_known_yet() { + when(context.getMetadataManager()).thenReturn(metadataManager); + when(metadataManager.getMetadata()).thenReturn(metadata); + when(metadata.getNodes()).thenReturn(Collections.emptyMap()); + + assertThat(handler.reresolvesNodeAddresses()).isTrue(); + } + // ---- Merge behavior tests ----------------------------------------------- @Test diff --git a/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapperTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/LoadBalancingPolicyWrapperTest.java index 89b36b9ee09..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/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); } From bee08377461b5206d33951af3a8443d0595fbb29 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Thu, 23 Jul 2026 23:25:38 +0200 Subject: [PATCH 6/9] docs: document contact-point DNS expansion in the upgrade guide (DRIVER-201) Add a 4.19.2.1 upgrade-guide entry covering the multi-address contact-point expansion, the deprecation of advanced.resolve-contact-points (now a no-op), and the fallback-to-original-contact-points default flip (false -> true) with how to restore the previous behavior. Co-Authored-By: Claude Opus 4.8 --- upgrade_guide/README.md | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) 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 From 46d1ae69b6c27201bc1d51403ca025061a970ea6 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Fri, 24 Jul 2026 00:05:13 +0200 Subject: [PATCH 7/9] fix: rotate SNI resolveAll() with a dedicated counter (DRIVER-201) resolveAll() shared the round-robin OFFSET counter with the deprecated resolve(). SniSslEngineFactory.newSslEngine() calls resolve() on every TLS connection, so a normal SNI-over-TLS connect advanced the counter twice and resolveAll()'s start index moved in steps of two -- collapsing rotation to a single proxy IP whenever the hostname resolved to an even number of A-records (index 0 in the common two-record case). Give resolveAll() its own RESOLVE_ALL_OFFSET so SSL engine setup no longer perturbs its rotation, and add a test that interleaves resolve() calls to lock this in. Co-Authored-By: Claude Opus 4.8 --- .../internal/core/metadata/SniEndPoint.java | 17 +++++++--- .../core/metadata/SniEndPointTest.java | 32 +++++++++++++++++++ 2 files changed, 45 insertions(+), 4 deletions(-) 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 acd3431580e..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 @@ -33,7 +33,14 @@ 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; @@ -86,9 +93,11 @@ public InetSocketAddress resolve() { * 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 the same round-robin {@link #OFFSET} - * counter as {@link #resolve()}. This spreads healthy connections across proxy IPs instead of - * always starting at the first one, preserving the previous load-balancing behavior. + * 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 @@ -112,7 +121,7 @@ public CompletionStage resolveAll(@NonNull Executor executor) { int start = (aRecords.length == 1) ? 0 - : OFFSET.getAndUpdate(x -> x == Integer.MAX_VALUE ? 0 : x + 1) + : 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++) { 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 index c7f5bb695e8..99672006747 100644 --- 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 @@ -98,4 +98,36 @@ public void resolve_all_returns_complete_and_rotated_candidate_order() { .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(); + } + } } From 2e9d9ff7dd38577691a2f15b6cf98c2834cf7575 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Fri, 24 Jul 2026 00:05:22 +0200 Subject: [PATCH 8/9] fix: aggregate multi-address connect failures; cover fallback with tests (DRIVER-201) When every candidate address from resolveAll() failed, ChannelFactory propagated only the last candidate's error and discarded the earlier ones (they were logged at DEBUG only). Attach the earlier failures as suppressed exceptions on the propagated error so the full picture is available for diagnosis. Add ChannelFactoryMultiAddressTest, which the existing single-address tests did not cover: first-candidate-fails/second-succeeds fallback, and the all-candidates-exhausted path (asserting the suppressed cause). Strengthen DefaultEndPointTest to assert resolveAll() expands to the complete DNS result set (compared against an independent InetAddress.getAllByName lookup), not just a non-empty subset. Co-Authored-By: Claude Opus 4.8 --- .../internal/core/channel/ChannelFactory.java | 23 +++- .../ChannelFactoryMultiAddressTest.java | 117 ++++++++++++++++++ .../core/metadata/DefaultEndPointTest.java | 23 +++- 3 files changed, 157 insertions(+), 6 deletions(-) create mode 100644 core/src/test/java/com/datastax/oss/driver/internal/core/channel/ChannelFactoryMultiAddressTest.java 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 d2e05ebef85..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,6 +58,7 @@ 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; @@ -290,7 +291,8 @@ private void connect( attemptedVersions, resultFuture, candidates, - 0); + 0, + new ArrayList<>()); }); } @@ -304,6 +306,11 @@ private void connect( * {@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, @@ -316,7 +323,8 @@ private void tryNextCandidate( List attemptedVersions, CompletableFuture resultFuture, SocketAddress[] candidates, - int index) { + int index, + List priorErrors) { SocketAddress candidate = candidates[index]; CompletableFuture perAddressFuture = new CompletableFuture<>(); @@ -342,6 +350,7 @@ private void tryNextCandidate( logPrefix, candidate, error.getMessage()); + priorErrors.add(error); tryNextCandidate( endPoint, shardingInfo, @@ -353,8 +362,16 @@ private void tryNextCandidate( attemptedVersions, resultFuture, candidates, - index + 1); + 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); } 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/metadata/DefaultEndPointTest.java b/core/src/test/java/com/datastax/oss/driver/internal/core/metadata/DefaultEndPointTest.java index 66b794cdcf8..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,8 +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 { @@ -69,17 +73,30 @@ public void resolve_all_returns_single_element_for_already_resolved_address() { } @Test - public void resolve_all_expands_unresolved_hostname_to_at_least_one_address() { - // localhost reliably resolves to at least 127.0.0.1 + 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(); - assertThat(all).isNotEmpty(); + + // 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 From bdd723280a3173b5cbd6eaff6504bd25a2c49e55 Mon Sep 17 00:00:00 2001 From: Mikita Hradovich Date: Fri, 24 Jul 2026 00:05:29 +0200 Subject: [PATCH 9/9] docs: fix contradictory resolve-contact-points doc; clarify reconnect option (DRIVER-201) The reference.conf entry for advanced.resolve-contact-points still described the old resolve-once/resolve-every-connection semantics as active above a "DEPRECATED: no effect" footer, contradicting itself. Lead with the deprecation and drop the stale active-voice prose. Clarify the CONTROL_CONNECTION_RECONNECT_CONTACT_POINTS Javadoc in TypedDriverOption: it appends the original (unresolved) contact points, expands them to all DNS IPs at connection time via EndPoint.resolveAll(), and is skipped for topology monitors that re-resolve node addresses themselves. Co-Authored-By: Claude Opus 4.8 --- .../api/core/config/TypedDriverOption.java | 10 +++++- core/src/main/resources/reference.conf | 33 +++++++------------ 2 files changed, 20 insertions(+), 23 deletions(-) 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 fa12f40fd3e..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 append the original contact points to the reconnection plan (defaults to true) */ + /** + * 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); diff --git a/core/src/main/resources/reference.conf b/core/src/main/resources/reference.conf index 24506b49bcc..b154f888481 100644 --- a/core/src/main/resources/reference.conf +++ b/core/src/main/resources/reference.conf @@ -1204,33 +1204,22 @@ 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 - # - # DEPRECATED: this option no longer has any effect. Contact points are always kept as unresolved - # hostnames and expanded to all their DNS-mapped IPs at connection time (see - # EndPoint.resolveAll()). It will be removed in a future release. advanced.resolve-contact-points = false advanced.protocol {