Propagate non-recoverable JSON conversion failures instead of demoting them to a generic 500

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

@@ -58,6 +58,7 @@
 import org.springframework.cloud.function.context.FunctionRegistry;
 import org.springframework.cloud.function.context.PostProcessingFunction;
 import org.springframework.cloud.function.context.config.KotlinLambdaToFunctionAutoConfiguration;
+import org.springframework.cloud.function.context.config.NonRecoverableConversionException;
 import org.springframework.cloud.function.context.config.RoutingFunction;
 import org.springframework.cloud.function.core.FunctionInvocationHelper;
 import org.springframework.cloud.function.json.JsonMapper;
@@ -1406,7 +1407,7 @@ private Object convertNonMessageInputIfNecessary(Type inputType, Object input,
 					}
 					catch (Exception e) {
 						if (failOnJsonError) {
-							throw e;
+							throw new NonRecoverableConversionException("Failed to convert JSON input to " + inputType, e);
 						}
 						if (logger.isDebugEnabled()) {
 							logger.debug("JSON conversion failed for '" + input + "' to " + inputType
@@ -1594,6 +1595,9 @@ else if (FunctionTypeUtils.isFlux(type) && publisher instanceof Mono) {
 						try {
 							return this.convertInputIfNecessary(v, actualType == null ? type : actualType);
 						}
+						catch (NonRecoverableConversionException e) {
+							throw e;
+						}
 						catch (Exception e) {
 							throw new IllegalStateException("Failed to convert input", e);
 						}
@@ -1602,6 +1606,9 @@ else if (FunctionTypeUtils.isFlux(type) && publisher instanceof Mono) {
 						try {
 							return this.convertInputIfNecessary(v, actualType == null ? type : actualType);
 						}
+						catch (NonRecoverableConversionException e) {
+							throw e;
+						}
 						catch (Exception e) {
 							throw new IllegalStateException("Failed to convert input", e);
 						}

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/NonRecoverableConversionException.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2020-present 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.function.context.config;
+
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.converter.MessageConversionException;
+
+/**
+ * Signals that input conversion failed in a way the caller has explicitly asked
+ * not to be silently swallowed (see {@code failOnJsonError} in
+ * {@link org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry}),
+ * as opposed to an ordinary conversion miss that a fallback (e.g. a
+ * {@code ConversionService}) may still recover from.
+ *
+ * <p>This is a distinct subtype specifically so that downstream error handling
+ * (e.g. a web framework's exception resolver) can recognize a definitive,
+ * non-recoverable conversion failure without needing to special-case every
+ * exception that conversion might otherwise throw.
+ *
+ * @author KOMUNE
+ * @since 5.0.3
+ */
+@SuppressWarnings("serial")
+public class NonRecoverableConversionException extends MessageConversionException {
+
+	public NonRecoverableConversionException(String description, @Nullable Throwable cause) {
+		super(description, cause);
+	}
+
+}

spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java

@@ -59,6 +59,7 @@
 import org.springframework.cloud.function.context.HybridFunctionalRegistrationTests.UppercaseFunction;
 import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
 import org.springframework.cloud.function.context.config.JsonMessageConverter;
+import org.springframework.cloud.function.context.config.NonRecoverableConversionException;
 import org.springframework.cloud.function.context.config.SmartCompositeMessageConverter;
 import org.springframework.cloud.function.json.GsonMapper;
 import org.springframework.cloud.function.json.JacksonMapper;
@@ -84,6 +85,7 @@
 
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 /**
  * @author Oleg Zhurakousky
@@ -496,6 +498,29 @@ public void testReactiveFunctionMessages() {
 		Assertions.assertThatIterable(blockFirst).isEqualTo(Arrays.asList("item1", "item2"));
 	}
 
+	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@Test
+	public void testReactivePojoFunctionPropagatesNonRecoverableConversionExceptionForMalformedJson() {
+		FunctionRegistration<ReactivePojoFunction> registration = new FunctionRegistration<>(new ReactivePojoFunction(), "reactivePojo")
+			.type(ReactivePojoFunction.class);
+
+		SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter,
+				new JacksonMapper(new ObjectMapper()));
+		catalog.register(registration);
+
+		Function lookedUpFunction = catalog.lookup("reactivePojo");
+
+		Flux<List<String>> result = (Flux<List<String>>) lookedUpFunction
+			.apply(Flux.just(MessageBuilder
+				// "name" is a JSON array here instead of a string - structurally mismatched, not just unparsable
+				.withPayload("[{\"name\":[\"not-a-string\"]}]")
+				.setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
+				.build()
+			));
+
+		assertThatThrownBy(result::blockFirst).isInstanceOf(NonRecoverableConversionException.class);
+	}
+
 	@SuppressWarnings({ "rawtypes", "unchecked" })
 	@Test
 	public void testWithCustomMessageConverter() {
@@ -833,6 +858,25 @@ public Flux<List<String>> apply(Flux<Message<List<Person>>> listFlux) {
 		}
 	}
 
+	/**
+	 * Same shape as {@link ReactiveFunction} but declared over the plain payload type
+	 * rather than {@code Message<List<Person>>} - this is the shape
+	 * {@link org.springframework.cloud.function.context.config.NonRecoverableConversionException}
+	 * is designed for: {@code SimpleFunctionRegistry#convertInputMessageIfNecessary}
+	 * deliberately falls back to the original, unconverted message when the target type
+	 * is itself {@code Message<T>} (to support legitimate no-conversion-needed cases like
+	 * KafkaNull), which means that shape never reaches the JSON-failure signal this type
+	 * introduces. A plain payload type does reach it.
+	 */
+	private static final class ReactivePojoFunction implements Function<Flux<List<Person>>, Flux<List<String>>> {
+
+		@Override
+		public Flux<List<String>> apply(Flux<List<Person>> listFlux) {
+			return listFlux
+				.map(lst -> lst.stream().map(Person::getName).collect(Collectors.toList()));
+		}
+	}
+
 	private static final class ReactiveMonoGreeter implements Supplier<Mono<Message<String>>> {
 
 		@Override