Refine behavior on error after response committed

If the response is set and we can't change the status through
ServerHttpResponse any more, allow the error signal to propagate and
let the individual server adapters handle it. Ultimately that should
result in closing the connection.

On Servlet containers, we check one last time if the response is
committed (we may not have filled the buffer). If not then save
the exception as a request attribute, dispatch, and re-throw it on the
container thread.

On Undertow access the connection and close it.

On Netty just let the error through to Reactor Netty.

Issue: SPR-16051

spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/RequestMappingExceptionHandlingIntegrationTests.java

@@ -18,8 +18,10 @@ package org.springframework.web.reactive.result.method.annotation;
 
 import java.io.IOException;
 
+import org.junit.Assume;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.context.ApplicationContext;
@@ -28,6 +30,7 @@ import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.ResponseEntity;
+import org.springframework.http.server.reactive.bootstrap.ReactorHttpServer;
 import org.springframework.web.bind.annotation.ExceptionHandler;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
@@ -53,30 +56,46 @@ public class RequestMappingExceptionHandlingIntegrationTests extends AbstractReq
 
 
 	@Test
-	public void controllerThrowingException() throws Exception {
-		String expected = "Recovered from error: State";
-		assertEquals(expected, performGet("/thrown-exception", new HttpHeaders(), String.class).getBody());
+	public void thrownException() throws Exception {
+		doTest("/thrown-exception", "Recovered from error: State");
 	}
 
 	@Test
-	public void controllerThrowingExceptionWithCause() throws Exception {
-		String expected = "Recovered from error: State";
-		assertEquals(expected, performGet("/thrown-exception-with-cause", new HttpHeaders(), String.class).getBody());
+	public void thrownExceptionWithCause() throws Exception {
+		doTest("/thrown-exception-with-cause", "Recovered from error: State");
 	}
 
 	@Test
-	public void controllerThrowingExceptionWithCauseToHandle() throws Exception {
-		String expected = "Recovered from error: IO";
-		String url = "/thrown-exception-with-cause-to-handle";
+	public void thrownExceptionWithCauseToHandle() throws Exception {
+		doTest("/thrown-exception-with-cause-to-handle", "Recovered from error: IO");
+	}
+
+	@Test
+	public void errorBeforeFirstItem() throws Exception {
+		doTest("/mono-error", "Recovered from error: Argument");
+	}
+
+	@Test // SPR-16051
+	public void exceptionAfterSeveralItems() throws Exception {
+
+		// TODO: uncomment and try after https://github.com/reactor/reactor-netty/issues/231
+		Assume.assumeFalse(server instanceof ReactorHttpServer);
+
+		try {
+			performGet("/SPR-16051", new HttpHeaders(), String.class).getBody();
+			fail();
+		}
+		catch (Throwable ex) {
+			String message = ex.getMessage();
+			assertNotNull(message);
+			assertTrue("Actual: " + message, message.startsWith("Error while extracting response"));
+		}
+	}
+
+	private void doTest(String url, String expected) throws Exception {
 		assertEquals(expected, performGet(url, new HttpHeaders(), String.class).getBody());
 	}
 
-	@Test
-	public void controllerReturnsMonoError() throws Exception {
-		String expected = "Recovered from error: Argument";
-		assertEquals(expected, performGet("/mono-error", new HttpHeaders(), String.class).getBody());
-	}
-
 
 	@Configuration
 	@EnableWebFlux
@@ -110,6 +129,18 @@ public class RequestMappingExceptionHandlingIntegrationTests extends AbstractReq
 			return Mono.error(new IllegalArgumentException("Argument"));
 		}
 
+		@GetMapping("/SPR-16051")
+		public Flux<String> errors() {
+			return Flux.range(1, 10000)
+					.map(i -> {
+						if (i == 1000) {
+							throw new RuntimeException("Random error");
+						}
+						return i + ". foo bar";
+					});
+		}
+
+
 		@ExceptionHandler
 		public Publisher<String> handleArgumentException(IOException ex) {
 			return Mono.just("Recovered from error: " + ex.getMessage());

spring-webflux/src/test/java/org/springframework/web/reactive/socket/AbstractWebSocketIntegrationTests.java

@@ -23,7 +23,6 @@ import java.net.URISyntaxException;
 import java.util.LinkedHashMap;
 import java.util.Map;
 
-import org.apache.tomcat.websocket.WsWebSocketContainer;
 import org.apache.tomcat.websocket.server.WsContextListener;
 import org.junit.After;
 import org.junit.Before;