Fix validateAfterInactivity on the async client

httpclient5-testing/src/test/java/org/apache/hc/client5/testing/TestValidateAfterInactivity.java

@@ -0,0 +1,292 @@
+/*
+ * ====================================================================
+ * 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
+ * <http://www.apache.org/>.
+ *
+ */
+
+package org.apache.hc.client5.testing;
+
+import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
+import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
+import org.apache.hc.client5.http.async.methods.SimpleRequestBuilder;
+import org.apache.hc.client5.http.classic.methods.HttpGet;
+import org.apache.hc.client5.http.config.ConnectionConfig;
+import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
+import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
+import org.apache.hc.client5.http.impl.classic.BasicHttpClientResponseHandler;
+import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
+import org.apache.hc.client5.http.impl.classic.HttpClients;
+import org.apache.hc.client5.http.impl.io.PoolingHttpClientConnectionManager;
+import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
+import org.apache.hc.client5.testing.TestValidateAfterInactivity.TcpReset;
+import org.apache.hc.core5.http.HttpHost;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketException;
+import java.net.StandardSocketOptions;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.apache.hc.core5.util.TimeValue.MAX_VALUE;
+import static org.apache.hc.core5.util.TimeValue.ZERO_MILLISECONDS;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Tests validateAfterInactivity behavior in both sync and async clients.
+ */
+class AbstractTestValidateAfterInactivity {
+    final AtomicInteger connectionsEstablished = new AtomicInteger(0);
+    final AtomicReference<SocketChannel> currentConnection = new AtomicReference<>();
+    volatile ServerSocketChannel serverSocket;
+    volatile int port;
+    volatile Thread serverThread;
+
+    @BeforeEach
+    void setup() throws Exception {
+        serverSocket = ServerSocketChannel.open().bind(new InetSocketAddress(0));
+        port = ((InetSocketAddress) serverSocket.getLocalAddress()).getPort();
+
+        serverThread = new Thread(this::runServer);
+        serverThread.setDaemon(true);
+        serverThread.start();
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        final SocketChannel socket = currentConnection.getAndSet(null);
+        if (socket != null) {
+            socket.close();
+        }
+
+        if (serverSocket != null) {
+            serverSocket.close();
+        }
+        if (serverThread != null) {
+            serverThread.interrupt();
+            serverThread.join(1000);
+        }
+    }
+
+    @Test
+    void testSyncClientWithStaleConnection() throws Exception {
+        try (final CloseableHttpClient client = syncClient(false)) {
+            sendPing(client);
+            sendPing(client);
+            assertEquals(1, connectionsEstablished.getAndSet(0));
+
+            closeServerEndOfConnection();
+
+            /*
+             * There are two things that can happen when reusing a stale connection.
+             *
+             * If we manage to send off the request and then read the end of the TCP stream, we will get a
+             * NoHttpResponseException. This corresponds to a TCP half-close: the server has only closed its own end of
+             * the connection, not the client's, so theoretically the client could send (and the server could swallow)
+             * an arbitrary amount of data.
+             *
+             * If we are unable to send the request at all, or we can only send part of the request, we will get a
+             * SocketException. This corresponds to a TCP reset; there's no such thing as "TCP half-reset."
+             */
+            final Class<? extends IOException> expectedException = this instanceof TcpReset ?
+                SocketException.class : IOException.class;
+            assertThrows(expectedException, () -> sendPing(client));
+            assertEquals(0, connectionsEstablished.get());
+        }
+    }
+
+    @Test
+    void testSyncClientWithValidateAfterInactivity() throws Exception {
+        try (final CloseableHttpClient client = syncClient(true)) {
+            sendPing(client);
+            sendPing(client);
+            assertEquals(1, connectionsEstablished.getAndSet(0));
+
+            closeServerEndOfConnection();
+
+            sendPing(client);
+            assertEquals(1, connectionsEstablished.get());
+        }
+    }
+
+    @Test
+    void testAsyncClientWithStaleConnection() throws Exception {
+        testAsyncClient(false);
+    }
+
+    @Test
+    void testAsyncClientWithValidateAfterInactivity() throws Exception {
+        testAsyncClient(true);
+    }
+
+    private void testAsyncClient(final boolean validateAfterInactivity) throws Exception {
+        try (final CloseableHttpAsyncClient client = asyncClient(validateAfterInactivity)) {
+            sendPing(client);
+            sendPing(client);
+            assertEquals(1, connectionsEstablished.getAndSet(0));
+
+            closeServerEndOfConnection();
+
+            sendPing(client);
+            assertEquals(1, connectionsEstablished.get());
+        }
+    }
+
+    protected void closeServerEndOfConnection() throws IOException, InterruptedException {
+        currentConnection.get().close();
+
+        // It is impossible to guarantee that a connection from the connection pool will not be closed mid-request.
+        // Even over localhost, closing a socket is inherently an asynchronous operation prone to race conditions.
+        // Not only do we need to see the TCP `FIN` or `RST` in time, but also the IOReactor (in the case of the async
+        // client) is asynchronously notified of the connection's closure; until processClosedSessions() runs, the stale
+        // connection will remain in the thread pool.
+        //
+        // These sorts of inherent race condition are unrelated to what is being asserted in these tests: it is always
+        // possible that a request will fail due to a closure race condition, but we want to ensure that connections
+        // are not reused when we know *from the beginning* that they are already closed.
+        Thread.sleep(50);
+    }
+
+    private void sendPing(final CloseableHttpClient client) throws URISyntaxException, IOException {
+        final HttpHost target = new HttpHost("localhost", port);
+        final URI requestUri = new URI("/ping");
+
+        final String response = client.execute(target, new HttpGet(requestUri), new BasicHttpClientResponseHandler());
+
+        assertEquals("OK", response);
+    }
+
+    private void sendPing(final CloseableHttpAsyncClient client) throws ExecutionException, InterruptedException {
+        final HttpHost target = new HttpHost("localhost", port);
+        final SimpleHttpRequest request = SimpleRequestBuilder.get().setHttpHost(target).setPath("/ping").build();
+
+        final SimpleHttpResponse response = client.execute(request, null).get();
+
+        assertEquals(200, response.getCode());
+    }
+
+    private void runServer() {
+        try {
+            while (!Thread.currentThread().isInterrupted() && serverSocket.isOpen()) {
+                final SocketChannel socketChannel = serverSocket.accept();
+                socketChannel.configureBlocking(true);
+                connectionsEstablished.incrementAndGet();
+                currentConnection.set(socketChannel);
+                handleConnection(socketChannel);
+            }
+        } catch (final IOException e) {
+            if (!Thread.currentThread().isInterrupted() && serverSocket.isOpen()) {
+                System.err.println("Server error: " + e.getClass() + e.getMessage());
+            }
+        }
+    }
+
+    private static void handleConnection(final SocketChannel socketChannel) throws IOException {
+        try {
+            final ByteBuffer buffer = ByteBuffer.allocate(4096);
+            while (socketChannel.isOpen()) {
+                buffer.clear();
+                final int bytesRead = socketChannel.read(buffer);
+                if (bytesRead <= 0) {
+                    return;
+                }
+
+                final String response = "HTTP/1.1 200 OK\r\n" +
+                    "Content-Type: text/plain\r\n" +
+                    "Content-Length: 2\r\n" +
+                    "Connection: keep-alive\r\n" +
+                    "\r\n" +
+                    "OK";
+
+                final ByteBuffer responseBuffer = ByteBuffer.wrap(response.getBytes(UTF_8));
+                while (responseBuffer.hasRemaining()) {
+                    socketChannel.write(responseBuffer);
+                }
+            }
+        } catch (final AsynchronousCloseException ignore) {
+            // Connection closure was initiated on the server's end
+        } catch (final IOException ex) {
+            if (ex.getMessage().startsWith("Connection reset")) {
+                System.err.println("Server saw connection closed by client");
+                return;
+            }
+            throw ex;
+        }
+    }
+
+    private CloseableHttpClient syncClient(final boolean validateAfterInactivity) {
+        final PoolingHttpClientConnectionManager connManager = new PoolingHttpClientConnectionManager();
+        connManager.setDefaultConnectionConfig(getConnectionConfig(validateAfterInactivity));
+        return HttpClients.custom()
+            .setConnectionManager(connManager)
+            .disableAutomaticRetries()
+            .build();
+    }
+
+    private CloseableHttpAsyncClient asyncClient(final boolean validateAfterInactivity) {
+        final PoolingAsyncClientConnectionManager connManager = new PoolingAsyncClientConnectionManager();
+        connManager.setDefaultConnectionConfig(getConnectionConfig(validateAfterInactivity));
+        final CloseableHttpAsyncClient client = HttpAsyncClients.custom()
+            .setConnectionManager(connManager)
+            .disableAutomaticRetries()
+            .build();
+        client.start();
+        return client;
+    }
+
+    private static ConnectionConfig getConnectionConfig(final boolean validateAfterInactivity) {
+        return ConnectionConfig.custom()
+            .setTimeToLive(MAX_VALUE)
+            .setValidateAfterInactivity(validateAfterInactivity ? ZERO_MILLISECONDS : MAX_VALUE)
+            .build();
+    }
+}
+
+public class TestValidateAfterInactivity {
+    @Nested
+    class TcpClose extends AbstractTestValidateAfterInactivity {
+    }
+
+    @Nested
+    class TcpReset extends AbstractTestValidateAfterInactivity {
+        @Override
+        protected void closeServerEndOfConnection() throws IOException, InterruptedException {
+            currentConnection.get().setOption(StandardSocketOptions.SO_LINGER, 0);
+            super.closeServerEndOfConnection();
+        }
+    }
+}

httpclient5/src/main/java/org/apache/hc/client5/http/impl/nio/PoolingAsyncClientConnectionManager.java

@@ -315,10 +315,6 @@ public void completed(final PoolEntry<HttpRoute, ManagedAsyncClientConnection> p
                                             })), Command.Priority.IMMEDIATE);
                                             return;
                                         }
-                                        if (LOG.isDebugEnabled()) {
-                                            LOG.debug("{} connection {} is closed", id, ConnPoolSupport.getId(connection));
-                                        }
-                                        poolEntry.discardConnection(CloseMode.IMMEDIATE);
                                     }
                                 }
                             }