GH-3610: Fix WebFluxMH for error handling

Fixes #3610

This fix changes the WebFluxRequestExecutingMessageHandler
The change specifically changes the contruction of the
WebClientResponseException to use the `create` factory method.

* Added changes and unit test to cover updated code

Updated code to use simplified exception construction that is more tolerant
  of larger payloads.
Updated unit tests to check for specific exception types.

* Corrected checkstyle errors

Corrected checkstyle error in imports.
Corrected whitespace between casts.
Corrected trailing whitespace.

* Added @author


**Cherry-pick to `5.4.x` & `5.3.x`**

Author: djgraff209

spring-integration-webflux/src/main/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,15 +17,13 @@
 package org.springframework.integration.webflux.outbound;
 
 import java.net.URI;
-import java.nio.charset.StandardCharsets;
 import java.util.Map;
 
 import org.reactivestreams.Publisher;
 
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.buffer.DataBuffer;
-import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.http.HttpEntity;
@@ -41,15 +39,13 @@
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
-import org.springframework.util.MimeType;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.reactive.function.BodyExtractor;
 import org.springframework.web.reactive.function.BodyExtractors;
 import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.reactive.function.BodyInserters;
 import org.springframework.web.reactive.function.client.ClientResponse;
 import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.reactive.function.client.WebClientResponseException;
 import org.springframework.web.util.DefaultUriBuilderFactory;
 
 import reactor.core.publisher.Flux;
@@ -62,6 +58,7 @@
  * @author Shiliang Li
  * @author Artem Bilan
  * @author Gary Russell
+ * @author David Graff
  *
  * @since 5.0
  *
@@ -299,32 +296,7 @@ private WebClient.RequestBodySpec createRequestBodySpec(Object uri, HttpMethod h
 
 	private Mono<ClientResponse> exchangeForResponseMono(WebClient.RequestBodySpec requestSpec) {
 		return requestSpec.retrieve()
-				.onStatus(HttpStatus::isError, response -> {
-					HttpStatus httpStatus = response.statusCode();
-					return response.body(BodyExtractors.toDataBuffers())
-							.reduce(DataBuffer::write)
-							.map(dataBuffer -> {
-								byte[] bytes = new byte[dataBuffer.readableByteCount()];
-								dataBuffer.read(bytes);
-								DataBufferUtils.release(dataBuffer);
-								return bytes;
-							})
-							.defaultIfEmpty(new byte[0])
-							.map(bodyBytes -> {
-										throw new WebClientResponseException(
-												"ClientResponse has erroneous status code: "
-														+ httpStatus.value() + " "
-														+ httpStatus.getReasonPhrase(),
-												httpStatus.value(),
-												httpStatus.getReasonPhrase(),
-												response.headers().asHttpHeaders(),
-												bodyBytes,
-												response.headers().contentType()
-														.map(MimeType::getCharset)
-														.orElse(StandardCharsets.ISO_8859_1));
-									}
-							);
-				})
+				.onStatus(HttpStatus::isError, ClientResponse::createException)
 				.toEntityList(DataBuffer.class)
 				.map((entity) ->
 						ClientResponse.create(entity.getStatusCode())

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-2021 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 
 import org.junit.jupiter.api.Test;
@@ -47,6 +48,7 @@
 /**
  * @author Shiliang Li
  * @author Artem Bilan
+ * @author David Graff
  *
  * @since 5.0
  */
@@ -108,6 +110,8 @@ void testReactiveErrorOneWay() {
 		assertThat(errorMessage).isNotNull();
 		assertThat(errorMessage).isInstanceOf(ErrorMessage.class);
 		Throwable throwable = (Throwable) errorMessage.getPayload();
+		assertThat(throwable).isInstanceOf(MessageHandlingException.class);
+		assertThat(throwable.getCause()).isInstanceOf(WebClientResponseException.Unauthorized.class);
 		assertThat(throwable.getMessage()).contains("401 Unauthorized");
 	}
 
@@ -173,7 +177,8 @@ void testServiceUnavailableWithoutBody() {
 		assertThat(payload).isInstanceOf(MessageHandlingException.class);
 
 		Exception exception = (Exception) payload;
-		assertThat(exception.getCause()).isInstanceOf(WebClientResponseException.class);
+		assertThat(exception).isInstanceOf(MessageHandlingException.class);
+		assertThat(exception.getCause()).isInstanceOf(WebClientResponseException.ServiceUnavailable.class);
 		assertThat(exception.getMessage()).contains("503 Service Unavailable");
 
 		Message<?> replyMessage = errorChannel.receive(10);
@@ -273,4 +278,57 @@ void testClientHttpResponseAsReply() {
 				.verifyComplete();
 	}
 
+
+	@Test
+	void testClientHttpResponseErrorAsReply() {
+		ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> {
+			response.setStatusCode(HttpStatus.NOT_FOUND);
+			response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
+
+			DataBufferFactory bufferFactory = response.bufferFactory();
+
+			Flux<DataBuffer> data =
+					Flux.just(
+						bufferFactory.wrap("{".getBytes(StandardCharsets.UTF_8)),
+						bufferFactory.wrap("  \"error\": \"Not Found\",".getBytes(StandardCharsets.UTF_8)),
+						bufferFactory.wrap("  \"message\": \"404 NOT_FOUND\",".getBytes(StandardCharsets.UTF_8)),
+						bufferFactory.wrap("  \"path\": \"/spring-integration\",".getBytes(StandardCharsets.UTF_8)),
+						bufferFactory.wrap("  \"status\": 404,".getBytes(StandardCharsets.UTF_8)),
+						bufferFactory.wrap("  \"timestamp\": \"1970-01-01T00:00:00.000+00:00\",".getBytes(StandardCharsets.UTF_8)),
+						bufferFactory.wrap("  \"trace\": \"some really\nlong\ntrace\",".getBytes(StandardCharsets.UTF_8)),
+						bufferFactory.wrap("}".getBytes(StandardCharsets.UTF_8))
+					);
+
+			return response.writeWith(data)
+					.then(Mono.defer(response::setComplete));
+		});
+
+		WebClient webClient = WebClient.builder()
+										.clientConnector(httpConnector)
+										.build();
+
+		String destinationUri = "https://www.springsource.org/spring-integration";
+		WebFluxRequestExecutingMessageHandler reactiveHandler =
+							new WebFluxRequestExecutingMessageHandler(destinationUri, webClient);
+
+		QueueChannel replyChannel = new QueueChannel();
+		QueueChannel errorChannel = new QueueChannel();
+		reactiveHandler.setOutputChannel(replyChannel);
+		reactiveHandler.setBodyExtractor(new ClientHttpResponseBodyExtractor());
+
+		final Message<?> message =
+							MessageBuilder.withPayload("hello, world")
+											.setErrorChannel(errorChannel)
+											.build();
+		reactiveHandler.handleMessage(message);
+
+		Message<?> errorMessage = errorChannel.receive(10_000);
+
+		assertThat(errorMessage).isNotNull();
+		assertThat(errorMessage).isInstanceOf(ErrorMessage.class);
+		final Throwable throwable = (Throwable) errorMessage.getPayload();
+		assertThat(throwable).isInstanceOf(MessageHandlingException.class);
+		assertThat(throwable.getCause()).isInstanceOf(WebClientResponseException.NotFound.class);
+		assertThat(throwable.getMessage()).contains("404 Not Found");
+	}
 }