diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/AsyncDnsResolver.java b/httpclient5/src/main/java/org/apache/hc/client5/http/AsyncDnsResolver.java new file mode 100644 index 0000000000..b1c7cd346c --- /dev/null +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/AsyncDnsResolver.java @@ -0,0 +1,69 @@ +/* + * ==================================================================== + * 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http; + +import org.apache.hc.core5.annotation.Experimental; +import org.apache.hc.core5.concurrent.Cancellable; + +/** + * Optional DNS resolver SPI capable of resolving a host's address families (A / AAAA) + * independently and asynchronously, reporting each family's result to + * {@link DnsResolutionCallback} as soon as it becomes available rather than blocking + * until every family has been queried (RFC 8305, Section 3). + * + *

This is a strictly optional extension of {@link DnsResolver}: when none is + * configured, resolution falls back to the plain, synchronous {@link DnsResolver}, + * which remains the primary resolution SPI. {@link DnsResolver} cannot itself + * implement this interface by running its blocking lookup on a separate thread: + * that would still serialize both families behind one blocking call and misrepresent + * it as an independent asynchronous resolution of each. Implement this interface only + * when the underlying resolution mechanism genuinely queries A and AAAA records + * independently, for example an asynchronous DNS client.

+ * + *

This SPI is considered experimental and may be discontinued in the future.

+ * + * @since 5.7 + */ +@Experimental +public interface AsyncDnsResolver { + + /** + * Starts resolving {@code host}. Implementations call {@link DnsResolutionCallback#resolved} + * once for each address family that resolves successfully, as soon as that family's + * result becomes available, independently of any other family still outstanding, and + * call {@link DnsResolutionCallback#failed} once for each family that fails or does not + * apply. Once every attempted family has either resolved or failed, implementations call + * {@link DnsResolutionCallback#completed()} exactly once. + * + * @param host the host name to resolve. + * @param port the port to attach to resolved addresses. + * @param callback receives each family's result as it becomes available. + * @return a handle that cancels any outstanding queries. + */ + Cancellable resolve(String host, int port, DnsResolutionCallback callback); + +} diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/DnsResolutionCallback.java b/httpclient5/src/main/java/org/apache/hc/client5/http/DnsResolutionCallback.java new file mode 100644 index 0000000000..49812ea857 --- /dev/null +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/DnsResolutionCallback.java @@ -0,0 +1,71 @@ +/* + * ==================================================================== + * 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. + * ==================================================================== + * + * This software consists of voluntary contributions made by many + * individuals on behalf of the Apache Software Foundation. For more + * information on the Apache Software Foundation, please see + * . + * + */ +package org.apache.hc.client5.http; + +import java.net.InetSocketAddress; +import java.util.List; + +import org.apache.hc.core5.annotation.Experimental; + +/** + * Receives the results of an {@link AsyncDnsResolver} resolution, one address family + * at a time, as each becomes available. + * + *

This SPI is considered experimental and may be discontinued in the future.

+ * + * @since 5.7 + */ +@Experimental +public interface DnsResolutionCallback { + + /** + * Called once a single address family (A or AAAA) has resolved successfully, as soon + * as that family's result is available and independently of any other family still + * being resolved. May be called more than once for the same resolution, once per + * family that resolves successfully; never called with an empty or {@code null} list. + * + * @param addresses the addresses resolved for one address family. + */ + void resolved(List addresses); + + /** + * Called once a single address family has failed to resolve or does not apply to + * the target host. May be called more than once, once per family that fails, + * including after {@link #resolved(List)} has already provided addresses for a + * different family. + * + * @param cause the reason this family could not be resolved. + */ + void failed(Exception cause); + + /** + * Called exactly once, after every attempted address family has either resolved + * or failed, to signal that no further results will arrive for this resolution. + */ + void completed(); + +} diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java index bb72be9f34..9851add192 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/DefaultAsyncClientConnectionOperator.java @@ -34,6 +34,7 @@ import java.nio.file.Path; import java.util.concurrent.Future; +import org.apache.hc.client5.http.AsyncDnsResolver; import org.apache.hc.client5.http.DnsResolver; import org.apache.hc.client5.http.SchemePortResolver; import org.apache.hc.client5.http.UnsupportedSchemeException; @@ -96,9 +97,22 @@ protected DefaultAsyncClientConnectionOperator( final DnsResolver dnsResolver, final boolean happyEyeballsEnabled, final TimeValue connectionAttemptDelay) { + this(tlsStrategyLookup, schemePortResolver, dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, null); + } + + /** + * @since 5.7 + */ + protected DefaultAsyncClientConnectionOperator( + final Lookup tlsStrategyLookup, + final SchemePortResolver schemePortResolver, + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay, + final AsyncDnsResolver asyncDnsResolver) { this.tlsStrategyLookup = Args.notNull(tlsStrategyLookup, "TLS strategy lookup"); this.schemePortResolver = schemePortResolver != null ? schemePortResolver : DefaultSchemePortResolver.INSTANCE; - this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay); + this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, asyncDnsResolver); } @Override diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java index 04ff6e09e1..9c42adde33 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeConnectionInitiator.java @@ -30,6 +30,7 @@ import java.net.SocketAddress; import java.util.concurrent.Future; +import org.apache.hc.client5.http.AsyncDnsResolver; import org.apache.hc.client5.http.DnsResolver; import org.apache.hc.core5.annotation.Contract; import org.apache.hc.core5.annotation.ThreadingBehavior; @@ -89,6 +90,35 @@ public MultihomeConnectionInitiator( this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay); } + /** + * Creates a multi-home connection initiator that, when {@code happyEyeballsEnabled} is + * {@code true} and {@code asyncDnsResolver} is not {@code null}, resolves each address + * family (A / AAAA) independently through {@code asyncDnsResolver} and races connection + * attempts against each family's addresses as soon as that family resolves (RFC 8305, + * Sections 3 and 8), rather than waiting for {@code dnsResolver} to resolve every family + * behind one blocking call. When {@code asyncDnsResolver} is {@code null}, behavior is + * identical to {@link #MultihomeConnectionInitiator(ConnectionInitiator, DnsResolver, boolean, TimeValue)}. + * + * @param connectionInitiator the connection initiator. + * @param dnsResolver the DNS resolver to use when {@code asyncDnsResolver} is {@code null}. + * @param happyEyeballsEnabled whether to race connection attempts instead of trying them + * sequentially. + * @param connectionAttemptDelay delay between the start of successive connection attempts; + * see {@link #MultihomeConnectionInitiator(ConnectionInitiator, DnsResolver, boolean, TimeValue)}. + * @param asyncDnsResolver optional resolver capable of resolving address families + * independently and asynchronously; may be {@code null}. + * @since 5.7 + */ + public MultihomeConnectionInitiator( + final ConnectionInitiator connectionInitiator, + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay, + final AsyncDnsResolver asyncDnsResolver) { + this.connectionInitiator = Args.notNull(connectionInitiator, "Connection initiator"); + this.sessionRequester = new MultihomeIOSessionRequester(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, asyncDnsResolver); + } + @Override public Future connect( final NamedEndpoint remoteEndpoint, diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java index 58c80cb13a..fc9d40ecc2 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequester.java @@ -28,6 +28,7 @@ package org.apache.hc.client5.http.impl.nio; import java.io.IOException; +import java.net.Inet6Address; import java.net.InetAddress; import java.net.InetSocketAddress; import java.net.SocketAddress; @@ -42,7 +43,9 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hc.client5.http.AsyncDnsResolver; import org.apache.hc.client5.http.ConnectExceptionSupport; +import org.apache.hc.client5.http.DnsResolutionCallback; import org.apache.hc.client5.http.DnsResolver; import org.apache.hc.client5.http.SystemDefaultDnsResolver; import org.apache.hc.core5.concurrent.Cancellable; @@ -74,6 +77,14 @@ final class MultihomeIOSessionRequester { */ private static final long MIN_CONNECTION_ATTEMPT_DELAY_MS = 10; + /** + * RFC 8305, Section 3: if the A (IPv4) response arrives before the AAAA (IPv6) one, wait + * this long for AAAA before proceeding with IPv4 alone, so the preferred family is not + * needlessly skipped because of reordering. "It is RECOMMENDED that the Resolution Delay + * be set to 50 milliseconds." No such wait applies when AAAA arrives first or already has. + */ + private static final TimeValue RESOLUTION_DELAY = TimeValue.ofMilliseconds(50); + /** * Lazily initialized, process-wide, single daemon-thread scheduler used solely to pace * staggered connection attempts (RFC 8305 connection racing). Shared across all instances @@ -88,6 +99,7 @@ private static final class LazyStaggerScheduler { private final DnsResolver dnsResolver; private final boolean happyEyeballsEnabled; private final TimeValue connectionAttemptDelay; + private final AsyncDnsResolver asyncDnsResolver; private final ScheduledExecutorService staggerScheduler; MultihomeIOSessionRequester(final DnsResolver dnsResolver) { @@ -98,7 +110,15 @@ private static final class LazyStaggerScheduler { final DnsResolver dnsResolver, final boolean happyEyeballsEnabled, final TimeValue connectionAttemptDelay) { - this(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, null); + this(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, null, null); + } + + MultihomeIOSessionRequester( + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay, + final AsyncDnsResolver asyncDnsResolver) { + this(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, asyncDnsResolver, null); } // Package-private: allows unit tests to supply a deterministic, synchronous scheduler @@ -108,6 +128,17 @@ private static final class LazyStaggerScheduler { final boolean happyEyeballsEnabled, final TimeValue connectionAttemptDelay, final ScheduledExecutorService staggerScheduler) { + this(dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, null, staggerScheduler); + } + + // Package-private: full constructor, letting unit tests control both the DNS + // resolution strategy and stagger scheduling. + MultihomeIOSessionRequester( + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay, + final AsyncDnsResolver asyncDnsResolver, + final ScheduledExecutorService staggerScheduler) { this.dnsResolver = dnsResolver != null ? dnsResolver : SystemDefaultDnsResolver.INSTANCE; this.happyEyeballsEnabled = happyEyeballsEnabled; final TimeValue delay = connectionAttemptDelay != null ? connectionAttemptDelay : DEFAULT_CONNECTION_ATTEMPT_DELAY; @@ -116,6 +147,7 @@ private static final class LazyStaggerScheduler { "Connection attempt delay must be at least %s ms (RFC 8305, Section 8)", MIN_CONNECTION_ATTEMPT_DELAY_MS); } this.connectionAttemptDelay = delay; + this.asyncDnsResolver = asyncDnsResolver; this.staggerScheduler = staggerScheduler; } @@ -174,6 +206,10 @@ public void cancelled() { LOG.debug("{} resolving remote address", remoteEndpoint.getHostName()); } + if (happyEyeballsEnabled && asyncDnsResolver != null) { + return connectWithAsyncResolution(connectionInitiator, remoteEndpoint, localAddress, connectTimeout, attachment, future); + } + final List remoteAddresses; try { remoteAddresses = dnsResolver.resolve(remoteEndpoint.getHostName(), remoteEndpoint.getPort()); @@ -280,6 +316,51 @@ public Future connect( return connect(connectionInitiator, remoteEndpoint, null, localAddress, connectTimeout, attachment, callback); } + /** + * Resolves {@code remoteEndpoint} via the configured {@link AsyncDnsResolver}, racing + * connection attempts against each address family's addresses as soon as that family + * resolves rather than waiting for every family to complete first (RFC 8305, Section 3: + * a Resolution Delay of 0ms trivially satisfies the "SHOULD NOT wait more than 50ms" + * recommendation, since connections begin immediately upon the first family's result). + * A family that resolves after the race has already been decided contributes nothing + * and is otherwise ignored. + */ + private Future connectWithAsyncResolution( + final ConnectionInitiator connectionInitiator, + final NamedEndpoint remoteEndpoint, + final SocketAddress localAddress, + final Timeout connectTimeout, + final Object attachment, + final ComplexFuture future) { + final Race race = new Race(connectionInitiator, remoteEndpoint, new ArrayList<>(), localAddress, + connectTimeout, attachment, future); + race.markResolutionPending(); + race.start(); + final Cancellable resolution = asyncDnsResolver.resolve( + remoteEndpoint.getHostName(), + remoteEndpoint.getPort(), + new DnsResolutionCallback() { + + @Override + public void resolved(final List addresses) { + race.onAddressesResolved(addresses); + } + + @Override + public void failed(final Exception cause) { + race.onResolutionFailed(cause); + } + + @Override + public void completed() { + race.onResolutionComplete(); + } + + }); + race.setResolutionCancellable(resolution); + return future; + } + /** * RFC 8305 connection racing: the first address is tried immediately; each subsequent * address is tried after {@link #connectionAttemptDelay} unless an outstanding attempt @@ -307,6 +388,19 @@ private final class Race implements Cancellable { private boolean cancelled; private ScheduledFuture pendingStagger; private long lastAttemptStartNanos = -1L; + // True once every address this race will ever see has been added, i.e. there is no + // DNS resolution still in flight that might contribute further addresses. Always + // true for a race constructed from a complete, pre-resolved address list; set to + // false by #markResolutionPending() for a race driven by progressive, asynchronous + // DNS resolution (see #connectWithAsyncResolution). + private boolean resolutionComplete = true; + private Cancellable resolution; + private Exception lastFailure; + // RFC 8305 Section 3 Resolution Delay bookkeeping: set only while waiting up to + // RESOLUTION_DELAY for AAAA after A arrived first. + private boolean ipv6Resolved; + private List bufferedIpv4; + private ScheduledFuture pendingResolutionDelay; Race( final ConnectionInitiator connectionInitiator, @@ -318,7 +412,7 @@ private final class Race implements Cancellable { final ComplexFuture future) { this.connectionInitiator = connectionInitiator; this.remoteEndpoint = remoteEndpoint; - this.remoteAddresses = remoteAddresses; + this.remoteAddresses = new ArrayList<>(remoteAddresses); this.localAddress = localAddress; this.connectTimeout = connectTimeout; this.attachment = attachment; @@ -328,6 +422,179 @@ private final class Race implements Cancellable { this.future.setDependency(this); } + void markResolutionPending() { + synchronized (lock) { + resolutionComplete = false; + } + } + + void setResolutionCancellable(final Cancellable resolution) { + final boolean alreadyOver; + synchronized (lock) { + alreadyOver = decided || cancelled; + if (!alreadyOver) { + this.resolution = resolution; + } + } + if (alreadyOver) { + resolution.cancel(); + } + } + + /** + * Addresses resolved for one address family. RFC 8305 Section 3: if AAAA (IPv6) + * arrives, it is merged in immediately. If A (IPv4) arrives and AAAA has neither + * already resolved nor is ruled out, it is buffered and connecting is held off for up + * to {@link #RESOLUTION_DELAY} to give the preferred family a chance to arrive first + * (Section 6: the two are then interleaved); if AAAA never arrives in time, or the + * resolution as a whole completes first, the buffered A addresses are merged in + * without further delay. + */ + void onAddressesResolved(final List addresses) { + synchronized (lock) { + if (decided || cancelled || addresses.isEmpty()) { + return; + } + final boolean isIpv6 = addresses.get(0).getAddress() instanceof Inet6Address; + if (isIpv6) { + ipv6Resolved = true; + if (pendingResolutionDelay != null) { + pendingResolutionDelay.cancel(false); + pendingResolutionDelay = null; + mergeAddressesLocked(addresses, true); + final List buffered = bufferedIpv4; + bufferedIpv4 = null; + if (buffered != null) { + mergeAddressesLocked(buffered, false); + } + } else { + mergeAddressesLocked(addresses, true); + } + return; + } + if (!ipv6Resolved && !resolutionComplete && pendingResolutionDelay == null && bufferedIpv4 == null) { + bufferedIpv4 = new ArrayList<>(addresses); + pendingResolutionDelay = staggerScheduler().schedule( + this::onResolutionDelayElapsed, RESOLUTION_DELAY.getDuration(), RESOLUTION_DELAY.getTimeUnit()); + return; + } + mergeAddressesLocked(addresses, false); + } + } + + private void onResolutionDelayElapsed() { + synchronized (lock) { + pendingResolutionDelay = null; + final List buffered = bufferedIpv4; + bufferedIpv4 = null; + if (decided || cancelled || buffered == null) { + return; + } + mergeAddressesLocked(buffered, false); + } + } + + /** + * Merges newly resolved addresses of one family into the race. Addresses already + * attempted (indices {@code < nextIndex}: in flight, succeeded or failed) are never + * touched. The still-pending suffix is rebuilt by splitting it back into its two + * families and re-interleaving with the new arrivals (RFC 8305 Section 6, IPv6 + * first), rather than simply appending the new addresses after whatever was already + * pending, which would let a newly arrived, preferred AAAA address end up queued + * behind an older, still-untried A address. + * + *

Must be called while holding {@code lock}.

+ */ + private void mergeAddressesLocked(final List newAddresses, final boolean newIsIpv6) { + final int attemptedCount = Math.min(nextIndex, remoteAddresses.size()); + final boolean hadNoNextAddress = attemptedCount >= remoteAddresses.size(); + final List pendingIpv6 = new ArrayList<>(); + final List pendingIpv4 = new ArrayList<>(); + for (int i = attemptedCount; i < remoteAddresses.size(); i++) { + final InetSocketAddress addr = remoteAddresses.get(i); + (addr.getAddress() instanceof Inet6Address ? pendingIpv6 : pendingIpv4).add(addr); + } + (newIsIpv6 ? pendingIpv6 : pendingIpv4).addAll(newAddresses); + + while (remoteAddresses.size() > attemptedCount) { + remoteAddresses.remove(remoteAddresses.size() - 1); + } + remoteAddresses.addAll(interleave(pendingIpv6, pendingIpv4)); + + if (!hadNoNextAddress || pendingStagger != null) { + // A next address was already known and either already started or already + // has a stagger pacing it: these new addresses simply join the pending queue + // in their RFC 8305 Section 6 order; nothing further to do right now. + return; + } + if (outstanding == 0) { + // Nothing in flight at all: start as promptly as RFC 8305 allows, exactly + // as a freshly resolved first address would (Resolution Delay, Section 3), + // while still respecting the minimum inter-attempt spacing (Section 5) in + // case an earlier attempt for a different family just failed. + scheduleNextPromptlyLocked(); + } else { + // An attempt is outstanding, but until now there was no further address to + // pace against; start the stagger clock now that one has arrived. + pendingStagger = staggerScheduler().schedule( + this::onStaggerElapsed, connectionAttemptDelay.getDuration(), connectionAttemptDelay.getTimeUnit()); + } + } + + /** Alternates between the two families (RFC 8305 Section 6), AAAA first. */ + private List interleave( + final List ipv6, final List ipv4) { + final List result = new ArrayList<>(ipv6.size() + ipv4.size()); + final int max = Math.max(ipv6.size(), ipv4.size()); + for (int i = 0; i < max; i++) { + if (i < ipv6.size()) { + result.add(ipv6.get(i)); + } + if (i < ipv4.size()) { + result.add(ipv4.get(i)); + } + } + return result; + } + + /** One address family failed to resolve; recorded as a candidate failure cause. */ + void onResolutionFailed(final Exception cause) { + synchronized (lock) { + lastFailure = cause; + } + } + + /** No further addresses will ever be added to this race. */ + void onResolutionComplete() { + boolean last = false; + Exception cause = null; + synchronized (lock) { + resolutionComplete = true; + if (pendingResolutionDelay != null) { + // AAAA will never arrive now: stop waiting for it and use IPv4 alone. + pendingResolutionDelay.cancel(false); + pendingResolutionDelay = null; + final List buffered = bufferedIpv4; + bufferedIpv4 = null; + if (buffered != null) { + mergeAddressesLocked(buffered, false); + } + } + if (!decided && !cancelled && outstanding == 0 && nextIndex >= remoteAddresses.size()) { + decided = true; + last = true; + cause = lastFailure != null ? lastFailure : new UnknownHostException(remoteEndpoint.getHostName()); + } + } + if (last) { + if (LOG.isDebugEnabled()) { + LOG.debug("{}:{} resolution complete with no further addresses to try ({})", + remoteEndpoint.getHostName(), remoteEndpoint.getPort(), cause.getClass()); + } + finish(cause); + } + } + void start() { synchronized (lock) { scheduleNextPromptlyLocked(); @@ -471,9 +738,10 @@ private void onAttemptFailed(final InetSocketAddress remoteAddress, final Except return; } cancelPendingStaggerLocked(); + lastFailure = cause; if (nextIndex < remoteAddresses.size()) { advance = true; - } else if (outstanding == 0) { + } else if (outstanding == 0 && resolutionComplete) { decided = true; last = true; } @@ -484,6 +752,9 @@ private void onAttemptFailed(final InetSocketAddress remoteAddress, final Except scheduleNextPromptlyLocked(); } } + // If neither advance nor last: every known address has been tried and failed, + // but DNS resolution is still pending and may yet contribute more (see + // #onAddressesResolved); wait for it rather than failing prematurely. if (last) { if (LOG.isDebugEnabled()) { LOG.debug("{}:{} connection to {} failed ({}); all addresses exhausted", @@ -513,7 +784,7 @@ private void onAttemptCancelled(final InetSocketAddress remoteAddress) { cancelPendingStaggerLocked(); if (nextIndex < remoteAddresses.size()) { advance = true; - } else if (outstanding == 0) { + } else if (outstanding == 0 && resolutionComplete) { decided = true; last = true; } @@ -532,7 +803,7 @@ private void onAttemptCancelled(final InetSocketAddress remoteAddress) { } private void finish(final Exception cause) { - if (cause instanceof IOException) { + if (cause instanceof IOException && !remoteAddresses.isEmpty()) { final InetAddress[] addresses = remoteAddresses.stream() .map(InetSocketAddress::getAddress) .toArray(InetAddress[]::new); @@ -545,17 +816,27 @@ private void finish(final Exception cause) { @Override public boolean cancel() { final List> toCancel; + final Cancellable resolutionToCancel; synchronized (lock) { if (decided || cancelled) { return false; } cancelled = true; cancelPendingStaggerLocked(); + if (pendingResolutionDelay != null) { + pendingResolutionDelay.cancel(false); + pendingResolutionDelay = null; + } + bufferedIpv4 = null; toCancel = new ArrayList<>(inFlight); + resolutionToCancel = resolution; } for (final Future attemptFuture : toCancel) { attemptFuture.cancel(true); } + if (resolutionToCancel != null) { + resolutionToCancel.cancel(); + } return true; } diff --git a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java index 17900c9017..77a3ea2e5f 100644 --- a/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java +++ b/httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManagerBuilder.java @@ -27,6 +27,7 @@ package org.apache.hc.client5.http.impl.nio; +import org.apache.hc.client5.http.AsyncDnsResolver; import org.apache.hc.client5.http.DnsResolver; import org.apache.hc.client5.http.HttpRoute; import org.apache.hc.client5.http.SchemePortResolver; @@ -60,6 +61,7 @@ public class PoolingAsyncClientConnectionManagerBuilder { private DnsResolver dnsResolver; private boolean happyEyeballsEnabled; private TimeValue connectionAttemptDelay; + private AsyncDnsResolver asyncDnsResolver; private PoolConcurrencyPolicy poolConcurrencyPolicy; private PoolReusePolicy poolReusePolicy; @@ -136,6 +138,25 @@ public final PoolingAsyncClientConnectionManagerBuilder setConnectionAttemptDela return this; } + /** + * Sets an optional {@link AsyncDnsResolver} used, when {@link #setHappyEyeballsEnabled(boolean)} + * is enabled, to resolve address families (A / AAAA) independently and race connection + * attempts against each family's addresses as soon as that family resolves (RFC 8305, + * Section 3), instead of waiting for {@link #setDnsResolver(DnsResolver)} to resolve every + * family behind one blocking call. Has no effect unless Happy Eyeballs racing is enabled; + * {@code null} (the default) falls back to {@link #setDnsResolver(DnsResolver)}. + * + * @param asyncDnsResolver the asynchronous DNS resolver, or {@code null} to use + * {@link #setDnsResolver(DnsResolver)}. + * @return this instance. + * @since 5.7 + */ + @Experimental + public final PoolingAsyncClientConnectionManagerBuilder setAsyncDnsResolver(final AsyncDnsResolver asyncDnsResolver) { + this.asyncDnsResolver = asyncDnsResolver; + return this; + } + /** * Sets {@link SchemePortResolver} instance. * @@ -317,6 +338,20 @@ protected AsyncClientConnectionOperator createConnectionOperator( final DnsResolver dnsResolver, final boolean happyEyeballsEnabled, final TimeValue connectionAttemptDelay) { + return createConnectionOperator(tlsStrategy, schemePortResolver, dnsResolver, happyEyeballsEnabled, connectionAttemptDelay, null); + } + + /** + * @since 5.7 + */ + @Internal + protected AsyncClientConnectionOperator createConnectionOperator( + final TlsStrategy tlsStrategy, + final SchemePortResolver schemePortResolver, + final DnsResolver dnsResolver, + final boolean happyEyeballsEnabled, + final TimeValue connectionAttemptDelay, + final AsyncDnsResolver asyncDnsResolver) { return new DefaultAsyncClientConnectionOperator( RegistryBuilder.create() .register(URIScheme.HTTPS.getId(), tlsStrategy) @@ -324,7 +359,8 @@ protected AsyncClientConnectionOperator createConnectionOperator( schemePortResolver, dnsResolver, happyEyeballsEnabled, - connectionAttemptDelay); + connectionAttemptDelay, + asyncDnsResolver); } public PoolingAsyncClientConnectionManager build() { @@ -340,7 +376,7 @@ public PoolingAsyncClientConnectionManager build() { } final PoolingAsyncClientConnectionManager poolingmgr = new PoolingAsyncClientConnectionManager( createConnectionOperator(tlsStrategyCopy, schemePortResolver, dnsResolver, - happyEyeballsEnabled, connectionAttemptDelay), + happyEyeballsEnabled, connectionAttemptDelay, asyncDnsResolver), poolConcurrencyPolicy, poolReusePolicy, null, diff --git a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java index 3997b2ee21..5e0d1a7b03 100644 --- a/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java +++ b/httpclient5/src/test/java/org/apache/hc/client5/http/impl/nio/MultihomeIOSessionRequesterTest.java @@ -41,9 +41,12 @@ import java.io.IOException; import java.net.InetAddress; import java.net.InetSocketAddress; +import java.net.SocketAddress; +import java.net.UnknownHostException; import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.Deque; import java.util.List; import java.util.concurrent.CancellationException; @@ -56,7 +59,10 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import org.apache.hc.client5.http.AsyncDnsResolver; +import org.apache.hc.client5.http.DnsResolutionCallback; import org.apache.hc.client5.http.DnsResolver; +import org.apache.hc.core5.concurrent.Cancellable; import org.apache.hc.core5.concurrent.FutureCallback; import org.apache.hc.core5.io.CloseMode; import org.apache.hc.core5.net.NamedEndpoint; @@ -213,6 +219,15 @@ private static List addresses(final int count) throws Excepti return out; } + private static List ipv6Addresses(final int count) throws Exception { + final List out = new ArrayList<>(count); + for (int i = 1; i <= count; i++) { + out.add(new InetSocketAddress(InetAddress.getByAddress( + new byte[]{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, (byte) i}), 8080)); + } + return out; + } + @SuppressWarnings("unchecked") private static FutureCallback[] captureCallbacks(final ConnectionInitiator connectionInitiator, final int expected) { final ArgumentCaptor captor = ArgumentCaptor.forClass(FutureCallback.class); @@ -568,15 +583,9 @@ void happyEyeballs_unexpectedCancellationOfNonLastAttempt_advancesToNextAddress( @Test void happyEyeballs_unexpectedCancellationOfLastAttempt_failsRatherThanHangingForever() throws Exception { - final List remoteAddresses = addresses(1); + final List twoAddresses = addresses(2); Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); Mockito.when(namedEndpoint.getPort()).thenReturn(8080); - Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); - - // A single address racing alone still goes through the Race (not the sequential path) - // when forced via the 4-arg constructor is not applicable here; use 2 addresses with - // the 2nd address also cancelled, so the race is driven to full exhaustion. - final List twoAddresses = addresses(2); Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(twoAddresses); final FakeScheduler scheduler = new FakeScheduler(); @@ -595,7 +604,6 @@ void happyEyeballs_unexpectedCancellationOfLastAttempt_failsRatherThanHangingFor assertFalse(future.isDone()); scheduler.fireNext(); assertTrue(future.isDone(), "the future must not be left pending forever after every attempt is cancelled"); - assertTrue(future.isCancelled() || future.isDone()); try { future.get(); fail("Expected an ExecutionException rather than an indefinitely pending future"); @@ -685,4 +693,334 @@ private static void await(final CountDownLatch latch) { Thread.currentThread().interrupt(); } } + + // ------------------------------------------------------------------------- + // Independent, asynchronous A/AAAA resolution (AsyncDnsResolver) + // ------------------------------------------------------------------------- + + /** Captures the {@link DnsResolutionCallback} passed to {@link AsyncDnsResolver#resolve}. */ + private static final class FakeAsyncDnsResolver implements AsyncDnsResolver { + DnsResolutionCallback callback; + final Cancellable resolutionHandle = Mockito.mock(Cancellable.class); + + @Override + public Cancellable resolve(final String host, final int port, final DnsResolutionCallback callback) { + this.callback = callback; + return resolutionHandle; + } + } + + @Test + void asyncDns_bothFamiliesResolveTogether_racesAllAddresses() throws Exception { + final List ipv6 = addresses(1); + final List ipv4 = addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final IOSession session = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(session); + return CompletableFuture.completedFuture(session); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertNotNull(asyncDnsResolver.callback, "resolver must have been invoked"); + asyncDnsResolver.callback.resolved(ipv6); + asyncDnsResolver.callback.completed(); + + assertTrue(future.isDone()); + assertEquals(session, future.get()); + Mockito.verify(connectionInitiator, Mockito.times(1)).connect(any(), any(), any(), any(), any(), any()); + assertTrue(ipv4.size() >= 0); // unused family in this scenario, kept for clarity of intent + } + + @Test + void asyncDns_secondFamilyArrivesWhileFirstStalls_mergedIntoOngoingRace() throws Exception { + // AAAA first, so merging is immediate and not subject to the Resolution Delay (that + // only applies when A arrives before AAAA); see the dedicated resolutionDelay_* tests. + final List firstFamily = ipv6Addresses(1); + final List secondFamily = addresses(2); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final IOSession winner = Mockito.mock(IOSession.class); + final Future firstAttempt = Mockito.mock(Future.class); + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> firstAttempt) // 1st address (first family): stalls + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(winner); + return CompletableFuture.completedFuture(winner); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + asyncDnsResolver.callback.resolved(firstFamily); + assertFalse(future.isDone(), "must not complete while the only known address is still stalled"); + assertEquals(0, scheduler.pendingCount(), "nothing to stagger towards until a 2nd address is known"); + + // The second family resolves while the first is still stalled: since no next address + // was known when the first attempt started, the stagger clock starts now instead. + asyncDnsResolver.callback.resolved(Collections.singletonList(secondFamily.get(0))); + asyncDnsResolver.callback.completed(); + assertEquals(1, scheduler.pendingCount(), "the newly arrived address must now be paced by a stagger"); + + scheduler.fireNext(); + assertTrue(future.isDone()); + assertEquals(winner, future.get()); + Mockito.verify(firstAttempt).cancel(true); + } + + @Test + void asyncDns_oneFamilyFailsOtherSucceeds_winsWithSucceedingFamily() throws Exception { + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final IOSession session = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(session); + return CompletableFuture.completedFuture(session); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + // AAAA fails (e.g. no IPv6 route), A succeeds: the resolution as a whole still works. + asyncDnsResolver.callback.failed(new UnknownHostException("no AAAA record")); + asyncDnsResolver.callback.resolved(addresses(1)); + asyncDnsResolver.callback.completed(); + + assertTrue(future.isDone()); + assertEquals(session, future.get()); + } + + @Test + void asyncDns_bothFamiliesFail_aggregatedFailure() { + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + final UnknownHostException ipv6Failure = new UnknownHostException("no AAAA record"); + asyncDnsResolver.callback.failed(ipv6Failure); + asyncDnsResolver.callback.failed(new UnknownHostException("no A record")); + asyncDnsResolver.callback.completed(); + + assertTrue(future.isDone()); + try { + future.get(); + fail("Expected ExecutionException"); + } catch (final InterruptedException | ExecutionException ex) { + assertInstanceOf(ExecutionException.class, ex); + } + Mockito.verifyNoInteractions(connectionInitiator); + } + + @Test + void asyncDns_cancellationBeforeResolution_cancelsOutstandingDnsQuery() { + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + future.cancel(true); + + Mockito.verify(asyncDnsResolver.resolutionHandle).cancel(); + assertTrue(future.isCancelled()); + + // A family that resolves after cancellation must be ignored, not start a connection. + asyncDnsResolver.callback.resolved(Collections.singletonList( + new InetSocketAddress(InetAddress.getLoopbackAddress(), 8080))); + Mockito.verifyNoInteractions(connectionInitiator); + } + + @Test + void asyncDns_notConfigured_fallsBackToSynchronousDnsResolver() throws Exception { + final List remoteAddresses = addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + Mockito.when(dnsResolver.resolve("somehost", 8080)).thenReturn(remoteAddresses); + + final IOSession session = Mockito.mock(IOSession.class); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(session); + return CompletableFuture.completedFuture(session); + }); + + // happyEyeballsEnabled=true but no AsyncDnsResolver configured (the default): must fall + // back to the plain, synchronous DnsResolver exactly as before this feature existed. + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250)); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + assertTrue(future.isDone()); + assertEquals(session, future.get()); + Mockito.verify(dnsResolver).resolve("somehost", 8080); + } + + @Test + void asyncDns_ipv4ArrivesFirst_waitsResolutionDelayForIpv6() throws Exception { + final List ipv4 = addresses(1); + final List ipv6 = ipv6Addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final IOSession session = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(session); + return CompletableFuture.completedFuture(session); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + // A arrives first: must wait (up to 50ms) for AAAA rather than connecting immediately. + asyncDnsResolver.callback.resolved(ipv4); + assertFalse(future.isDone()); + Mockito.verifyNoInteractions(connectionInitiator); + + // AAAA arrives within the delay: it must be tried first (RFC 8305 interleaving). + asyncDnsResolver.callback.resolved(ipv6); + asyncDnsResolver.callback.completed(); + + assertTrue(future.isDone()); + assertEquals(session, future.get()); + final ArgumentCaptor addressCaptor = ArgumentCaptor.forClass(SocketAddress.class); + Mockito.verify(connectionInitiator).connect(any(), addressCaptor.capture(), any(), any(), any(), any()); + assertEquals(ipv6.get(0), addressCaptor.getValue(), "AAAA must be tried before the buffered A address"); + } + + @Test + void asyncDns_ipv4ArrivesFirst_aaaaNeverArrives_usesIpv4AfterDelayElapses() throws Exception { + final List ipv4 = addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final IOSession session = Mockito.mock(IOSession.class); + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> { + final FutureCallback cb = invocation.getArgument(5); + cb.completed(session); + return CompletableFuture.completedFuture(session); + }); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + final Future future = requester.connect( + connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + asyncDnsResolver.callback.resolved(ipv4); + assertFalse(future.isDone()); + assertEquals(1, scheduler.pendingCount(), "the Resolution Delay timer must be pending"); + + scheduler.fireNext(); // simulate the 50ms Resolution Delay elapsing with no AAAA + assertTrue(future.isDone()); + assertEquals(session, future.get()); + } + + @Test + void asyncDns_ipv4ArrivesFirst_aaaaArrivesAfterRacingAlreadyStarted_mergedBehindInFlightAddress() throws Exception { + final List ipv4 = addresses(2); + final List ipv6 = ipv6Addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final Future firstAttempt = Mockito.mock(Future.class); + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> firstAttempt); // stalls + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + requester.connect(connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + asyncDnsResolver.callback.resolved(ipv4); + scheduler.fireNext(); // Resolution Delay elapses with no AAAA: racing starts with IPv4 alone + Mockito.verify(connectionInitiator, Mockito.times(1)).connect(any(), any(), any(), any(), any(), any()); + + // AAAA now arrives late, after racing has already started on the first IPv4 address. + // It must be queued ahead of the still-pending (not yet attempted) 2nd IPv4 address, + // not appended after it. + asyncDnsResolver.callback.resolved(ipv6); + asyncDnsResolver.callback.completed(); + + scheduler.fireNext(); // fire the stagger for the 2nd (now re-ordered) pending attempt + final ArgumentCaptor addressCaptor = ArgumentCaptor.forClass(SocketAddress.class); + Mockito.verify(connectionInitiator, Mockito.times(2)).connect(any(), addressCaptor.capture(), any(), any(), any(), any()); + assertEquals(ipv6.get(0), addressCaptor.getAllValues().get(1), + "the late AAAA address must be tried before the already-pending, not-yet-attempted 2nd A address"); + } + + @Test + void asyncDns_lateIpv6Result_doesNotDisturbAlreadyAttemptedAddress() throws Exception { + final List ipv4First = addresses(1); + final List ipv6Late = ipv6Addresses(1); + Mockito.when(namedEndpoint.getHostName()).thenReturn("somehost"); + Mockito.when(namedEndpoint.getPort()).thenReturn(8080); + + final Future firstAttempt = Mockito.mock(Future.class); + final FakeScheduler scheduler = new FakeScheduler(); + final FakeAsyncDnsResolver asyncDnsResolver = new FakeAsyncDnsResolver(); + Mockito.when(connectionInitiator.connect(any(), any(), any(), any(), any(), any())) + .thenAnswer(invocation -> firstAttempt); + + final MultihomeIOSessionRequester requester = + new MultihomeIOSessionRequester(dnsResolver, true, TimeValue.ofMilliseconds(250), asyncDnsResolver, scheduler.mock); + requester.connect(connectionInitiator, namedEndpoint, null, Timeout.ofSeconds(5), null, null); + + asyncDnsResolver.callback.resolved(ipv4First); + scheduler.fireNext(); + final ArgumentCaptor firstCaptor = ArgumentCaptor.forClass(SocketAddress.class); + Mockito.verify(connectionInitiator).connect(any(), firstCaptor.capture(), any(), any(), any(), any()); + assertEquals(ipv4First.get(0), firstCaptor.getValue()); + + // A late AAAA address arrives after the only known address is already attempting; + // the already-in-flight attempt must be left untouched. + asyncDnsResolver.callback.resolved(ipv6Late); + asyncDnsResolver.callback.completed(); + Mockito.verify(connectionInitiator, Mockito.times(1)).connect(any(), any(), any(), any(), any(), any()); + } }