Dispatch in UndertowHttpHandlerAdapter

This ensures that the reactive handling of the request is dispatched
from the Undertow IO thread, marking the exchange as async rather than
ending it once the Undertow `handleRequest` method returns.

See gh-33885
Closes gh-33969

spring-web/src/main/java/org/springframework/http/server/reactive/UndertowHttpHandlerAdapter.java

@@ -66,6 +66,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
 
 	@Override
 	public void handleRequest(HttpServerExchange exchange) {
+		exchange.dispatch(() -> {
 			UndertowServerHttpRequest request = null;
 			try {
 				request = new UndertowServerHttpRequest(exchange, getDataBufferFactory());
@@ -85,6 +86,7 @@ public class UndertowHttpHandlerAdapter implements io.undertow.server.HttpHandle
 
 			HandlerResultSubscriber resultSubscriber = new HandlerResultSubscriber(exchange, request);
 			this.httpHandler.handle(request, response).subscribe(resultSubscriber);
+		});
 	}
 
 

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

@@ -17,6 +17,7 @@
 package org.springframework.web.reactive.result.method.annotation;
 
 import java.nio.ByteBuffer;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -331,6 +332,17 @@ class RequestMappingMessageConversionIntegrationTests extends AbstractRequestMap
 		assertThat(performPost("/person-transform/flux", JSON, req, JSON, PERSON_LIST).getBody()).isEqualTo(res);
 	}
 
+	@ParameterizedHttpServerTest // see gh-33885
+	void personTransformWithFluxDelayed(HttpServer httpServer) throws Exception {
+		startServer(httpServer);
+
+		List<?> req = asList(new Person("Robert"), new Person("Marie"));
+		List<?> res = asList(new Person("ROBERT"), new Person("MARIE"));
+		assertThat(performPost("/person-transform/flux-delayed", JSON, req, JSON, PERSON_LIST))
+				.satisfies(r -> assertThat(r.getBody()).isEqualTo(res))
+				.satisfies(r -> assertThat(r.getHeaders().getContentLength()).isNotZero());
+	}
+
 	@ParameterizedHttpServerTest
 	void personTransformWithObservable(HttpServer httpServer) throws Exception {
 		startServer(httpServer);
@@ -632,6 +644,11 @@ class RequestMappingMessageConversionIntegrationTests extends AbstractRequestMap
 			return persons.map(person -> new Person(person.getName().toUpperCase()));
 		}
 
+		@PostMapping("/flux-delayed")
+		Flux<Person> transformDelayed(@RequestBody Flux<Person> persons) {
+			return transformFlux(persons).delayElements(Duration.ofMillis(10));
+		}
+
 		@PostMapping("/observable")
 		Observable<Person> transformObservable(@RequestBody Observable<Person> persons) {
 			return persons.map(person -> new Person(person.getName().toUpperCase()));