Fix missing notifyRemoteAsyncErrors http config wiring

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/AbstractTest.java

@@ -58,7 +58,12 @@ public class AbstractTest
 
     protected void start(Handler handler) throws Exception
     {
-        HTTP2CServerConnectionFactory connectionFactory = new HTTP2CServerConnectionFactory(new HttpConfiguration());
+        start(handler, new HttpConfiguration());
+    }
+
+    protected void start(Handler handler, HttpConfiguration httpConfiguration) throws Exception
+    {
+        HTTP2CServerConnectionFactory connectionFactory = new HTTP2CServerConnectionFactory(httpConfiguration);
         connectionFactory.setInitialSessionRecvWindow(FlowControlStrategy.DEFAULT_WINDOW_SIZE);
         connectionFactory.setInitialStreamRecvWindow(FlowControlStrategy.DEFAULT_WINDOW_SIZE);
         prepareServer(connectionFactory);

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/AsyncIOTest.java

@@ -17,22 +17,33 @@
 import java.nio.ByteBuffer;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.eclipse.jetty.http.HttpFields;
 import org.eclipse.jetty.http.MetaData;
+import org.eclipse.jetty.http2.ErrorCode;
 import org.eclipse.jetty.http2.api.Session;
 import org.eclipse.jetty.http2.api.Stream;
 import org.eclipse.jetty.http2.frames.DataFrame;
 import org.eclipse.jetty.http2.frames.HeadersFrame;
+import org.eclipse.jetty.http2.frames.ResetFrame;
 import org.eclipse.jetty.io.Content;
+import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.server.Handler;
+import org.eclipse.jetty.server.HttpConfiguration;
 import org.eclipse.jetty.server.Request;
 import org.eclipse.jetty.server.Response;
+import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.FuturePromise;
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.nullValue;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
@@ -241,6 +252,67 @@ public void onClosed(Stream stream)
 */
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testClientResetRemoteErrorNotification(boolean notify) throws Exception
+    {
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicReference<Response> responseRef = new AtomicReference<>();
+        AtomicReference<Throwable> failureRef = new AtomicReference<>();
+        HttpConfiguration httpConfiguration = new HttpConfiguration();
+        httpConfiguration.setNotifyRemoteAsyncErrors(notify);
+        start(new Handler.Abstract()
+        {
+            @Override
+            public boolean handle(Request request, Response response, Callback callback)
+            {
+                request.addFailureListener(failureRef::set);
+                responseRef.set(response);
+                latch.countDown();
+                return true;
+            }
+        }, httpConfiguration);
+
+        Session session = newClientSession(new Session.Listener() {});
+        MetaData.Request metaData = newRequest("GET", HttpFields.EMPTY);
+        HeadersFrame frame = new HeadersFrame(metaData, null, true);
+        FuturePromise<Stream> promise = new FuturePromise<>();
+        session.newStream(frame, promise, null);
+        Stream stream = promise.get(5, TimeUnit.SECONDS);
+
+        // Wait for the server to be idle.
+        assertTrue(latch.await(5, TimeUnit.SECONDS));
+        sleep(500);
+
+        stream.reset(new ResetFrame(stream.getId(), ErrorCode.CANCEL_STREAM_ERROR.code), Callback.NOOP);
+
+        if (notify)
+            // Wait for the reset to be notified to the failure listener.
+            await().atMost(5, TimeUnit.SECONDS).until(failureRef::get, instanceOf(EofException.class));
+        else
+            // Wait for the reset to NOT be notified to the failure listener.
+            await().atMost(5, TimeUnit.SECONDS).during(1, TimeUnit.SECONDS).until(failureRef::get, nullValue());
+
+        // Assert that writing to the response fails.
+        var cb = new Callback()
+        {
+            private Throwable failure = null;
+
+            @Override
+            public void failed(Throwable x)
+            {
+                failure = x;
+            }
+
+            Throwable failure()
+            {
+                return failure;
+            }
+        };
+        responseRef.get().write(true, BufferUtil.EMPTY_BUFFER, cb);
+        await().atMost(5, TimeUnit.SECONDS).until(cb::failure, instanceOf(EofException.class));
+    }
+
     private static void sleep(long ms) throws InterruptedIOException
     {
         try

jetty-core/jetty-http2/jetty-http2-tests/src/test/java/org/eclipse/jetty/http2/tests/AsyncServletTest.java

@@ -154,58 +154,6 @@ public void test()
 //        assertTrue(clientLatch.await(2 * idleTimeout, TimeUnit.MILLISECONDS));
 //    }
 //
-//    @Test
-//    public void testStartAsyncThenClientResetWithoutRemoteErrorNotification() throws Exception
-//    {
-//        HttpConfiguration httpConfiguration = new HttpConfiguration();
-//        httpConfiguration.setNotifyRemoteAsyncErrors(false);
-//        prepareServer(new HTTP2ServerConnectionFactory(httpConfiguration));
-//        ServletContextHandler context = new ServletContextHandler(server, "/");
-//        AtomicReference<AsyncContext> asyncContextRef = new AtomicReference<>();
-//        CountDownLatch latch = new CountDownLatch(1);
-//        context.addServlet(new ServletHolder(new HttpServlet()
-//        {
-//            @Override
-//            protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException
-//            {
-//                AsyncContext asyncContext = request.startAsync();
-//                asyncContext.setTimeout(0);
-//                asyncContextRef.set(asyncContext);
-//                latch.countDown();
-//            }
-//        }), servletPath + "/*");
-//        server.start();
-//
-//        prepareClient();
-//        client.start();
-//        Session session = newClient(new Session.Listener() {});
-//        MetaData.Request metaData = newRequest("GET", HttpFields.EMPTY);
-//        HeadersFrame frame = new HeadersFrame(metaData, null, true);
-//        FuturePromise<Stream> promise = new FuturePromise<>();
-//        session.newStream(frame, promise, null);
-//        Stream stream = promise.get(5, TimeUnit.SECONDS);
-//
-//        // Wait for the server to be in ASYNC_WAIT.
-//        assertTrue(latch.await(5, TimeUnit.SECONDS));
-//        sleep(500);
-//
-//        stream.reset(new ResetFrame(stream.getId(), ErrorCode.CANCEL_STREAM_ERROR.code), Callback.NOOP);
-//
-//        // Wait for the reset to be processed by the server.
-//        sleep(500);
-//
-//        AsyncContext asyncContext = asyncContextRef.get();
-//        ServletResponse response = asyncContext.getResponse();
-//        ServletOutputStream output = response.getOutputStream();
-//
-//        assertThrows(IOException.class,
-//            () ->
-//            {
-//                // Large writes or explicit flush() must
-//                // fail because the stream has been reset.
-//                output.flush();
-//            });
-//    }
 //
 //    @Test
 //    public void testStartAsyncThenServerSessionIdleTimeout() throws Exception

jetty-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java

@@ -437,7 +437,7 @@ public Runnable onFailure(Throwable x)
                 // Notify the failure listeners only once.
                 Consumer<Throwable> onFailure = _onFailure;
                 _onFailure = null;
-                Runnable invokeOnFailureListeners = onFailure == null ? null : () ->
+                Runnable invokeOnFailureListeners = onFailure == null || !getHttpConfiguration().isNotifyRemoteAsyncErrors() ? null : () ->
                 {
                     try
                     {

jetty-ee10/jetty-ee10-servlet/src/main/java/org/eclipse/jetty/ee10/servlet/HttpOutput.java

@@ -676,7 +676,9 @@ public void flush() throws IOException
             catch (Throwable t)
             {
                 onWriteComplete(false, t);
-                throw t;
+                if (t instanceof IOException)
+                    throw t;
+                throw new IOException(t);
             }
         }
     }

jetty-ee10/jetty-ee10-servlet/src/main/java/org/eclipse/jetty/ee10/servlet/ServletChannelState.java

@@ -572,7 +572,8 @@ public void startAsync(AsyncContextEvent event)
             if (!_failureListener)
             {
                 _failureListener = true;
-                _servletChannel.getRequest().addFailureListener(this::asyncError);
+                if (_servletChannel.getHttpConfiguration().isNotifyRemoteAsyncErrors())
+                    _servletChannel.getRequest().addFailureListener(this::asyncError);
             }
             _requestState = RequestState.ASYNC;
             _event = event;

jetty-ee10/jetty-ee10-tests/jetty-ee10-test-client-transports/src/test/java/org/eclipse/jetty/ee10/test/client/transport/AbstractTest.java

@@ -14,13 +14,15 @@
 package org.eclipse.jetty.ee10.test.client.transport;
 
 import java.io.InputStream;
+import java.net.InetSocketAddress;
 import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.security.KeyStore;
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 import jakarta.servlet.http.HttpServlet;
@@ -32,7 +34,13 @@
 import org.eclipse.jetty.ee10.servlet.ServletHolder;
 import org.eclipse.jetty.fcgi.client.transport.HttpClientTransportOverFCGI;
 import org.eclipse.jetty.fcgi.server.ServerFCGIConnectionFactory;
+import org.eclipse.jetty.http.HostPortHttpField;
+import org.eclipse.jetty.http.HttpFields;
+import org.eclipse.jetty.http.HttpScheme;
+import org.eclipse.jetty.http.HttpVersion;
+import org.eclipse.jetty.http.MetaData;
 import org.eclipse.jetty.http2.HTTP2Cipher;
+import org.eclipse.jetty.http2.api.Session;
 import org.eclipse.jetty.http2.client.HTTP2Client;
 import org.eclipse.jetty.http2.client.transport.HttpClientTransportOverHTTP2;
 import org.eclipse.jetty.http2.server.AbstractHTTP2ServerConnectionFactory;
@@ -58,6 +66,7 @@
 import org.eclipse.jetty.server.SslConnectionFactory;
 import org.eclipse.jetty.toolchain.test.jupiter.WorkDir;
 import org.eclipse.jetty.toolchain.test.jupiter.WorkDirExtension;
+import org.eclipse.jetty.util.FuturePromise;
 import org.eclipse.jetty.util.SocketAddressResolver;
 import org.eclipse.jetty.util.component.LifeCycle;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
@@ -79,6 +88,7 @@ public class AbstractTest
     protected AbstractConnector connector;
     protected ServletContextHandler servletContextHandler;
     protected HttpClient client;
+    protected HTTP2Client http2Client;
 
     public static Collection<Transport> transports()
     {
@@ -189,6 +199,29 @@ protected void startClient(Transport transport, Consumer<HttpClient> consumer) t
         client.start();
     }
 
+    protected Session newHttp2ClientSession(Session.Listener listener) throws Exception
+    {
+        String host = "localhost";
+        int port = ((NetworkConnector)connector).getLocalPort();
+        InetSocketAddress address = new InetSocketAddress(host, port);
+        FuturePromise<Session> promise = new FuturePromise<>();
+        http2Client.connect(address, listener, promise);
+        return promise.get(5, TimeUnit.SECONDS);
+    }
+
+    protected MetaData.Request newRequest(String method, HttpFields fields)
+    {
+        return newRequest(method, "/", fields);
+    }
+
+    protected MetaData.Request newRequest(String method, String path, HttpFields fields)
+    {
+        String host = "localhost";
+        int port = ((NetworkConnector)connector).getLocalPort();
+        String authority = host + ":" + port;
+        return new MetaData.Request(method, HttpScheme.HTTP.asString(), new HostPortHttpField(authority), path, HttpVersion.HTTP_2, fields, -1);
+    }
+
     public AbstractConnector newConnector(Transport transport, Server server)
     {
         return switch (transport)
@@ -262,7 +295,7 @@ protected HttpClientTransport newHttpClientTransport(Transport transport) throws
                 ClientConnector clientConnector = new ClientConnector();
                 clientConnector.setSelectors(1);
                 clientConnector.setSslContextFactory(newSslContextFactoryClient());
-                HTTP2Client http2Client = new HTTP2Client(clientConnector);
+                http2Client = new HTTP2Client(clientConnector);
                 yield new HttpClientTransportOverHTTP2(http2Client);
             }
             case H3 ->