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 ->