Speed up testRespondAfterClose (#117969)

Since #106511 this test takes 30s because it waits for the client to
time out and close the connection before allowing the transport to fully
shut down. This commit reinstates the previous behaviour of closing
connections quickly, and tests both client-triggered and
server-triggered connection closure.

Author: DaveCTurner

modules/transport-netty4/src/test/java/org/elasticsearch/http/netty4/Netty4HttpServerTransportTests.java

@@ -40,6 +40,7 @@
 import io.netty.handler.codec.http.HttpUtil;
 import io.netty.handler.codec.http.HttpVersion;
 
+import org.apache.http.ConnectionClosedException;
 import org.apache.http.HttpHost;
 import org.apache.lucene.util.SetOnce;
 import org.elasticsearch.ElasticsearchException;
@@ -48,6 +49,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.bulk.IncrementalBulkService;
 import org.elasticsearch.action.support.ActionTestUtils;
+import org.elasticsearch.action.support.PlainActionFuture;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.client.Request;
 import org.elasticsearch.client.RestClient;
@@ -100,6 +102,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
@@ -110,6 +113,7 @@
 import static com.carrotsearch.randomizedtesting.RandomizedTest.getRandom;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ALLOW_ORIGIN;
 import static org.elasticsearch.http.HttpTransportSettings.SETTING_CORS_ENABLED;
+import static org.elasticsearch.http.HttpTransportSettings.SETTING_HTTP_SERVER_SHUTDOWN_GRACE_PERIOD;
 import static org.elasticsearch.rest.RestStatus.BAD_REQUEST;
 import static org.elasticsearch.rest.RestStatus.OK;
 import static org.elasticsearch.rest.RestStatus.UNAUTHORIZED;
@@ -1039,8 +1043,16 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
         }
     }
 
-    public void testRespondAfterClose() throws Exception {
-        final String url = "/thing";
+    public void testRespondAfterServiceCloseWithClientCancel() throws Exception {
+        runRespondAfterServiceCloseTest(true);
+    }
+
+    public void testRespondAfterServiceCloseWithServerCancel() throws Exception {
+        runRespondAfterServiceCloseTest(false);
+    }
+
+    private void runRespondAfterServiceCloseTest(boolean clientCancel) throws Exception {
+        final String url = "/" + randomIdentifier();
         final CountDownLatch responseReleasedLatch = new CountDownLatch(1);
         final SubscribableListener<Void> transportClosedFuture = new SubscribableListener<>();
         final CountDownLatch handlingRequestLatch = new CountDownLatch(1);
@@ -1066,7 +1078,9 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
 
         try (
             Netty4HttpServerTransport transport = new Netty4HttpServerTransport(
-                Settings.EMPTY,
+                clientCancel
+                    ? Settings.EMPTY
+                    : Settings.builder().put(SETTING_HTTP_SERVER_SHUTDOWN_GRACE_PERIOD.getKey(), TimeValue.timeValueMillis(1)).build(),
                 networkService,
                 threadPool,
                 xContentRegistry(),
@@ -1082,11 +1096,24 @@ public void dispatchBadRequest(final RestChannel channel, final ThreadContext th
             transport.start();
             final var address = randomFrom(transport.boundAddress().boundAddresses()).address();
             try (var client = RestClient.builder(new HttpHost(address.getAddress(), address.getPort())).build()) {
-                client.performRequestAsync(new Request("GET", url), ActionTestUtils.wrapAsRestResponseListener(ActionListener.noop()));
+                final var responseExceptionFuture = new PlainActionFuture<Exception>();
+                final var cancellable = client.performRequestAsync(
+                    new Request("GET", url),
+                    ActionTestUtils.wrapAsRestResponseListener(ActionTestUtils.assertNoSuccessListener(responseExceptionFuture::onResponse))
+                );
                 safeAwait(handlingRequestLatch);
+                if (clientCancel) {
+                    threadPool.generic().execute(cancellable::cancel);
+                }
                 transport.close();
                 transportClosedFuture.onResponse(null);
                 safeAwait(responseReleasedLatch);
+                final var responseException = safeGet(responseExceptionFuture);
+                if (clientCancel) {
+                    assertThat(responseException, instanceOf(CancellationException.class));
+                } else {
+                    assertThat(responseException, instanceOf(ConnectionClosedException.class));
+                }
             }
         }
     }