diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java index e2ce54ae3..efe01ff64 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java +++ b/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); } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/NonRecoverableConversionException.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/NonRecoverableConversionException.java new file mode 100644 index 000000000..f0e2d045e --- /dev/null +++ b/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. + * + *

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); + } + +} diff --git a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java b/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java index 64a273556..9a044b35a 100644 --- a/spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistryTests.java +++ b/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 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> result = (Flux>) 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> apply(Flux>> listFlux) { } } + /** + * Same shape as {@link ReactiveFunction} but declared over the plain payload type + * rather than {@code Message>} - 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} (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>> { + + @Override + public Flux> apply(Flux> listFlux) { + return listFlux + .map(lst -> lst.stream().map(Person::getName).collect(Collectors.toList())); + } + } + private static final class ReactiveMonoGreeter implements Supplier>> { @Override