GH-SCST-2235 Fix Partitioning issue with FunctionAroundWrapper"

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

@@ -37,13 +37,20 @@ public abstract class FunctionAroundWrapper implements BiFunction<Object, Functi
 	@SuppressWarnings("unchecked")
 	@Override
 	public final Object apply(Object input, FunctionInvocationWrapper targetFunction) {
+		boolean isSkipOutputConversion = targetFunction.isSkipOutputConversion();
+		targetFunction.setSkipOutputConversion(true);
+		Object result = null;
 		if (input instanceof Message) {
-			return this.doApply((Message<byte[]>) input, targetFunction);
+			result = this.doApply((Message<byte[]>) input, targetFunction);
 		}
 		else if (targetFunction.isSupplier() && !targetFunction.isOutputTypePublisher()) {
-			return this.doApply(null, targetFunction);
+			result = this.doApply(null, targetFunction);
 		}
-		return targetFunction.apply(input);
+		else {
+			result = targetFunction.apply(input);
+		}
+		targetFunction.setSkipOutputConversion(isSkipOutputConversion);
+		return result;
 	}
 
 	protected abstract Object doApply(Message<byte[]> input, FunctionInvocationWrapper targetFunction);

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

@@ -390,6 +390,7 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 		private Function<Object, Object> enhancer;
 
 		FunctionInvocationWrapper(FunctionInvocationWrapper function) {
+			this.expectedOutputContentType = function.expectedOutputContentType;
 			this.skipOutputConversion = function.skipOutputConversion;
 			this.skipInputConversion = function.skipInputConversion;
 			this.target = function.target;
@@ -407,6 +408,11 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			this.message = this.inputType != null && FunctionTypeUtils.isMessage(this.inputType);
 		}
 
+		public boolean isSkipOutputConversion() {
+			return skipOutputConversion;
+		}
+
+
 		public boolean isPrototype() {
 			return this.isPrototype();
 		}