Fix generics for customizeMonoReply()

Related to: https://stackoverflow.com/questions/68637283/how-to-customize-response-in-spring-integration-using-webflux-when-a-specific-er

The function provided for the `ConsumerEndpointSpec.customizeMonoReply()` may convert
incoming value to something else.
With wildcards it cannot be compiled without casting.

* Add `<T, V>` generic arg for the `customizeMonoReply()` to conform in and out types carrying.
* Modify `WebFluxDslTests` to demonstrate the problem and confirm the fix

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

Author: artembilan

spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java

@@ -201,12 +201,16 @@ public S transactional(boolean handleMessageAdvice) {
 	/**
 	 * Specify a {@link BiFunction} for customizing {@link Mono} replies via {@link ReactiveRequestHandlerAdvice}.
 	 * @param replyCustomizer the {@link BiFunction} to propagate into {@link ReactiveRequestHandlerAdvice}.
+	 * @param <T> inbound reply payload.
+	 * @param <V> outbound reply payload.
 	 * @return the spec.
 	 * @since 5.3
 	 * @see ReactiveRequestHandlerAdvice
 	 */
-	public S customizeMonoReply(BiFunction<Message<?>, Mono<?>, Publisher<?>> replyCustomizer) {
-		return advice(new ReactiveRequestHandlerAdvice(replyCustomizer));
+	@SuppressWarnings({ "unchecked", "rawtypes"})
+	public <T, V> S customizeMonoReply(BiFunction<Message<?>, Mono<T>, Publisher<V>> replyCustomizer) {
+		return advice(new ReactiveRequestHandlerAdvice(
+				(BiFunction<Message<?>, Mono<?>, Publisher<?>>) (BiFunction) replyCustomizer));
 	}
 
 	/**

spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java

@@ -88,6 +88,7 @@
 import org.springframework.web.reactive.config.EnableWebFlux;
 import org.springframework.web.reactive.config.WebFluxConfigurer;
 import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.reactive.function.client.WebClientResponseException;
 import org.springframework.web.util.UriComponentsBuilder;
 
 import reactor.core.publisher.Flux;
@@ -418,8 +419,12 @@ public IntegrationFlow webFluxFlowWithReplyPayloadToFlux() {
 									.id("webFluxWithReplyPayloadToFlux")
 									.customizeMonoReply(
 											(message, mono) ->
-													mono.timeout(Duration.ofMillis(100))
-															.retry()));
+													mono.
+															timeout(Duration.ofMillis(100))
+															.retry()
+															.onErrorResume(
+																	WebClientResponseException.NotFound.class,
+																	ex -> Mono.just("Not Found"))));
 		}
 
 		@Bean