Auto-detect native transport by default when its library is available

client/src/main/java/org/asynchttpclient/AsyncHttpClientConfig.java

@@ -436,6 +436,13 @@ default AddressResolverGroup<InetSocketAddress> getAddressResolverGroup() {
         return null;
     }
 
+    /**
+     * Whether a native transport was explicitly requested. Note that {@code false} no longer forces NIO:
+     * a native transport is auto-selected whenever its library is on the classpath. Set
+     * {@code -Dio.netty.transport.noNative=true} to force NIO.
+     *
+     * @return true if a native transport was explicitly requested
+     */
     boolean isUseNativeTransport();
 
     boolean isUseOnlyEpollNativeTransport();

client/src/main/java/org/asynchttpclient/DefaultAsyncHttpClientConfig.java

@@ -1643,6 +1643,15 @@ public Builder setAddressResolverGroup(@Nullable AddressResolverGroup<InetSocket
             return this;
         }
 
+        /**
+         * Requests a native transport, failing back to NIO if none is available on this platform.
+         * <p>
+         * Passing {@code false} does not force NIO: when unset, a native transport is still auto-selected
+         * if its library is on the classpath. Use {@code -Dio.netty.transport.noNative=true} to force NIO.
+         *
+         * @param useNativeTransport whether to explicitly request a native transport
+         * @return the same builder instance
+         */
         public Builder setUseNativeTransport(boolean useNativeTransport) {
             this.useNativeTransport = useNativeTransport;
             return this;

client/src/main/java/org/asynchttpclient/netty/channel/ChannelManager.java

@@ -129,6 +129,8 @@ public class ChannelManager {
     // Guards the one-time WARN emitted when a native transport was requested but is unavailable and we
     // fall back to NIO. Logged once per JVM to avoid spamming logs when many clients are created.
     private static final AtomicBoolean NATIVE_FALLBACK_WARNED = new AtomicBoolean();
+    // Guards the one-time WARN emitted when io_uring allocation fails and we fall back to epoll.
+    private static final AtomicBoolean IO_URING_FALLBACK_WARNED = new AtomicBoolean();
     private final AsyncHttpClientConfig config;
     private final SslEngineFactory sslEngineFactory;
     private final EventLoopGroup eventLoopGroup;
@@ -203,30 +205,47 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) {
         ThreadFactory threadFactory = config.getThreadFactory() != null ? config.getThreadFactory() : new DefaultThreadFactory(config.getThreadPoolName());
         allowReleaseEventLoopGroup = config.getEventLoopGroup() == null;
         TransportFactory<? extends Channel, ? extends EventLoopGroup> transportFactory;
+        EventLoopGroup localEventLoopGroup;
 
         if (allowReleaseEventLoopGroup) {
             if (config.isUseNativeTransport()) {
                 transportFactory = getNativeTransportFactory(config);
             } else {
-                transportFactory = NioTransportFactory.INSTANCE;
+                transportFactory = autoSelectTransportFactory();
+            }
+            try {
+                localEventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory);
+            } catch (Throwable t) {
+                if (transportFactory instanceof IoUringTransportFactory && EpollTransportFactory.isAvailable()) {
+                    if (IO_URING_FALLBACK_WARNED.compareAndSet(false, true)) {
+                        LOGGER.warn("io_uring event loop group creation failed ({}); falling back to epoll. "
+                                + "io_uring rings count against RLIMIT_MEMLOCK (~76 KB per io thread, "
+                                + "{} threads requested); raise 'ulimit -l' to use io_uring.",
+                                t, config.getIoThreadsCount());
+                    }
+                    transportFactory = new EpollTransportFactory();
+                    localEventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory);
+                } else {
+                    throw t;
+                }
             }
-            eventLoopGroup = transportFactory.newEventLoopGroup(config.getIoThreadsCount(), threadFactory);
         } else {
-            eventLoopGroup = config.getEventLoopGroup();
+            localEventLoopGroup = config.getEventLoopGroup();
 
-            if (eventLoopGroup instanceof NioEventLoopGroup) {
+            if (localEventLoopGroup instanceof NioEventLoopGroup) {
                 transportFactory = NioTransportFactory.INSTANCE;
-            } else if (isInstanceof(eventLoopGroup, "io.netty.channel.epoll.EpollEventLoopGroup")) {
+            } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.epoll.EpollEventLoopGroup")) {
                 transportFactory = new EpollTransportFactory();
-            } else if (isInstanceof(eventLoopGroup, "io.netty.channel.kqueue.KQueueEventLoopGroup")) {
+            } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.kqueue.KQueueEventLoopGroup")) {
                 transportFactory = new KQueueTransportFactory();
-            } else if (isInstanceof(eventLoopGroup, "io.netty.channel.uring.IOUringEventLoopGroup")) {
+            } else if (isInstanceof(localEventLoopGroup, "io.netty.channel.uring.IOUringEventLoopGroup")) {
                 transportFactory = new IoUringTransportFactory();
             } else {
-                throw new IllegalArgumentException("Unknown event loop group " + eventLoopGroup.getClass().getSimpleName());
+                throw new IllegalArgumentException("Unknown event loop group " + localEventLoopGroup.getClass().getSimpleName());
             }
         }
 
+        this.eventLoopGroup = localEventLoopGroup;
         channelOptions = buildChannelOptions(config);
         httpBootstrap = newBootstrap(transportFactory, eventLoopGroup);
         wsBootstrap = newBootstrap(transportFactory, eventLoopGroup);
@@ -267,6 +286,24 @@ public ChannelManager(final AsyncHttpClientConfig config, Timer nettyTimer) {
         return NioTransportFactory.INSTANCE;
     }
 
+    // Default when useNativeTransport is unset: native transport if its lib is on the classpath (silently),
+    // else NIO. Use -Dio.netty.transport.noNative=true to force NIO.
+    private static TransportFactory<? extends Channel, ? extends EventLoopGroup> autoSelectTransportFactory() {
+        if (PlatformDependent.isOsx()) {
+            if (KQueueTransportFactory.isAvailable()) {
+                return new KQueueTransportFactory();
+            }
+        } else if (!PlatformDependent.isWindows()) {
+            // Prefer epoll: io_uring needs RLIMIT_MEMLOCK (often constrained in CI/containers).
+            if (EpollTransportFactory.isAvailable()) {
+                return new EpollTransportFactory();
+            } else if (IoUringTransportFactory.isAvailable()) {
+                return new IoUringTransportFactory();
+            }
+        }
+        return NioTransportFactory.INSTANCE;
+    }
+
     public static boolean isSslHandlerConfigured(ChannelPipeline pipeline) {
         return pipeline.get(SSL_HANDLER) != null;
     }

client/src/main/java/org/asynchttpclient/netty/channel/NettyChannelConnector.java

@@ -23,6 +23,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.net.ConnectException;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.util.List;
@@ -118,9 +119,20 @@ public void onFailure(Channel channel, Throwable t) {
                         if (retry) {
                             connect(bootstrap, connectListener);
                         } else {
-                            connectListener.onFailure(channel, t);
+                            connectListener.onFailure(channel, annotateConnectException(t, remoteAddress));
                         }
                     }
                 });
     }
+
+    private Throwable annotateConnectException(Throwable t, InetSocketAddress remoteAddress) {
+        if (t instanceof ConnectException) {
+            return t;  // Already has proper type; preserve for retry predicates
+        }
+        String address = remoteAddress.toString();
+        String message = t.getMessage();
+        ConnectException annotated = new ConnectException((message != null ? message : t.getClass().getSimpleName()) + ": " + address);
+        annotated.initCause(t);
+        return annotated;
+    }
 }

client/src/main/java/org/asynchttpclient/netty/future/StackTraceInspector.java

@@ -56,8 +56,14 @@ public static boolean recoverOnNettyDisconnectException(Throwable t) {
 
     public static boolean recoverOnReadOrWriteException(Throwable t) {
         while (true) {
-            if (t instanceof IOException && "Connection reset by peer".equalsIgnoreCase(t.getMessage())) {
-                return true;
+            // Native transports (epoll, io_uring, kqueue) report resets as NativeIoException with the
+            // strerror text baked into the message, e.g. "recvAddress(..) failed with error(-104):
+            // Connection reset by peer". Modern JDKs drop the "by peer" suffix, hence the substring match.
+            if (t instanceof IOException) {
+                String msg = t.getMessage();
+                if (msg != null && msg.contains("Connection reset")) {
+                    return true;
+                }
             }
 
             try {

client/src/main/java/org/asynchttpclient/request/body/multipart/part/InputStreamMultipartPart.java

@@ -33,6 +33,7 @@ public class InputStreamMultipartPart extends FileLikeMultipartPart<InputStreamP
     private long position;
     private ByteBuffer buffer;
     private ReadableByteChannel channel;
+    private boolean sourceExhausted;
 
     public InputStreamMultipartPart(InputStreamPart part, byte[] boundary) {
         super(part, boundary);
@@ -77,17 +78,46 @@ protected long transferContentTo(WritableByteChannel target) throws IOException
         ByteBuffer buffer = getBuffer();
 
         int transferred = 0;
-        int read = channel.read(buffer);
 
-        if (read > 0) {
+        if (buffer.position() > 0) {
             buffer.flip();
-            while (buffer.hasRemaining()) {
-                transferred += target.write(buffer);
+            int written = target.write(buffer);
+            if (written > 0) {
+                transferred += written;
+                position += written;
             }
             buffer.compact();
-            position += transferred;
+            if (written == 0) {
+                slowTarget = true;
+                return 0;
+            }
         }
-        if (position == getContentLength() || read < 0) {
+
+        // Stop reading once the declared length is accounted for: a socket-backed stream has nothing more
+        // to give and would block that read until the request times out.
+        long contentLength = getContentLength();
+        boolean allBytesInHand = contentLength >= 0 && position + buffer.position() >= contentLength;
+
+        if (!sourceExhausted && !allBytesInHand) {
+            int read = channel.read(buffer);
+            if (read > 0) {
+                buffer.flip();
+                int written = target.write(buffer);
+                if (written > 0) {
+                    transferred += written;
+                    position += written;
+                }
+                buffer.compact();
+                if (written == 0) {
+                    slowTarget = true;
+                }
+            } else if (read < 0) {
+                sourceExhausted = true;
+            }
+        }
+
+        boolean allDeclaredBytesWritten = contentLength >= 0 && position >= contentLength;
+        if ((sourceExhausted || allDeclaredBytesWritten) && buffer.position() == 0) {
             state = MultipartState.POST_CONTENT;
             if (channel.isOpen()) {
                 channel.close();

client/src/test/java/org/asynchttpclient/DefaultAsyncHttpClientTest.java

@@ -105,6 +105,21 @@ public void testNativeTransportFallsBackToNioWhenNativeUnavailable() throws IOEx
         }
     }
 
+    @RepeatedIfExceptionsTest(repeats = 5)
+    @EnabledOnOs(OS.LINUX)
+    public void testAutoSelectsNativeTransportByDefaultWhenAvailable() throws IOException {
+        AsyncHttpClientConfig config = config().build();
+        try (DefaultAsyncHttpClient client = (DefaultAsyncHttpClient) asyncHttpClient(config)) {
+            EventLoopGroup group = client.channelManager().getEventLoopGroup();
+            boolean nativeAvailable = Epoll.isAvailable() || IoUring.isAvailable();
+            if (nativeAvailable) {
+                assertFalse(group instanceof NioEventLoopGroup, "default config must auto-select native transport when available");
+            } else {
+                assertInstanceOf(NioEventLoopGroup.class, group, "no native transport available -> NIO");
+            }
+        }
+    }
+
     @RepeatedIfExceptionsTest(repeats = 5)
     public void testUseOnlyEpollNativeTransportButNativeTransportIsDisabled() {
         assertThrows(IllegalArgumentException.class, () -> config().setUseNativeTransport(false).setUseOnlyEpollNativeTransport(true).build());