Remove attempt to convert incoming type to target type from Requestprocessor

This fix effectively forces all type conversion happen in function catalog

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

@@ -21,6 +21,7 @@ import java.lang.reflect.GenericArrayType;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -41,6 +42,7 @@ import org.aopalliance.intercept.MethodInterceptor;
 import org.aopalliance.intercept.MethodInvocation;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.json.JSONObject;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -793,13 +795,19 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 					}
 				}
 				else if (rawType instanceof Class<?>) { // see AWS adapter with WildardTypeImpl and Azure with Voids
-					try {
+					if (this.isJson(value)) {
-						convertedValue = conversionService.convert(value, (Class<?>) rawType);
+						convertedValue = messageConverter
-					}
-					catch (Exception e) {
-						if (value instanceof String || value instanceof byte[]) {
-							convertedValue = messageConverter
 								.fromMessage(new GenericMessage<Object>(value), (Class<?>) rawType);
+					}
+					else {
+						try {
+							convertedValue = conversionService.convert(value, (Class<?>) rawType);
+						}
+						catch (Exception e) {
+							if (value instanceof String || value instanceof byte[]) {
+								convertedValue = messageConverter
+									.fromMessage(new GenericMessage<Object>(value), (Class<?>) rawType);
+							}
 						}
 					}
 				}
@@ -813,6 +821,22 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			return convertedValue;
 		}
 
+		private boolean isJson(Object value) {
+			String v = value instanceof byte[]
+					? new String((byte[]) value, StandardCharsets.UTF_8)
+							: (value instanceof String ? (String) value : null);
+			if (v != null) {
+				try {
+					new JSONObject(v);
+					return true;
+				}
+				catch (Exception ex) {
+					// ignore
+				}
+			}
+			return false;
+		}
+
 		private boolean messageNeedsConversion(Type rawType, Message<?> message) {
 			Boolean skipConversion = message.getHeaders().containsKey(FunctionProperties.SKIP_CONVERSION_HEADER)
 				? message.getHeaders().get(FunctionProperties.SKIP_CONVERSION_HEADER, Boolean.class)

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java

@@ -127,7 +127,9 @@ public class RequestProcessor {
 
 	public Mono<ResponseEntity<?>> post(FunctionWrapper wrapper,
 			ServerWebExchange exchange) {
-		Mono<ResponseEntity<?>> responseEntity = Mono.from(body(wrapper.handler(), exchange))
+		Mono<ResponseEntity<?>> responseEntity = Mono
+				.from(body(wrapper.handler(), exchange))
+				.doOnError(e -> logger.error("Failed to generate POST input for function: " + wrapper.function, e))
 				.flatMap(body -> response(wrapper, body, false));
 
 		return responseEntity;
@@ -207,7 +209,6 @@ public class RequestProcessor {
 			boolean stream) {
 
 		Function function = wrapper.function();
-
 		Flux<?> flux;
 		if (body != null) {
 			if (Collection.class
@@ -258,7 +259,7 @@ public class RequestProcessor {
 			}
 			else {
 				result = Flux.from((Publisher) result);
-				logger.debug("Handled POST with function");
+				logger.debug("Handled POST with function: " + function);
 				if (stream) {
 					responseEntityMono = stream(wrapper, result);
 				}
@@ -344,7 +345,12 @@ public class RequestProcessor {
 	private Publisher<?> body(Object handler, ServerWebExchange exchange) {
 		ResolvableType elementType = ResolvableType
 				.forClass(this.inspector.getInputType(handler));
+
+		// we effectively delegate type conversion to FunctionCatalog
+		elementType = ResolvableType.forClass(String.class);
+
 		ResolvableType actualType = elementType;
+
 		Class<?> resolvedType = elementType.resolve();
 		ReactiveAdapter adapter = (resolvedType != null
 				? getAdapterRegistry().getAdapter(resolvedType) : null);
@@ -386,8 +392,7 @@ public class RequestProcessor {
 				else {
 					// Single-value (with or without reactive type wrapper)
 					if (logger.isDebugEnabled()) {
-						logger.debug(
+						logger.debug(exchange.getLogPrefix() + "0..1 [" + elementType + "]");
-								exchange.getLogPrefix() + "0..1 [" + elementType + "]");
 					}
 					Mono<?> mono = reader.readMono(actualType, elementType, request,
 							response, readHints).doOnNext(v -> {