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 ac8f54878..fa5673e4e 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 @@ -1214,7 +1214,8 @@ else if (this.skipInputConversion) { } else if (input instanceof Message) { input = this.filterOutHeaders((Message) input); - if (((Message) input).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) { + if (this.isInputTypeMessage() + && ((Message) input).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) { return input; } 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 47f9b1be1..b9afda93c 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 @@ -69,6 +69,7 @@ import org.springframework.core.ResolvableType; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.DefaultConversionService; +import org.springframework.kafka.support.KafkaNull; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -157,6 +158,62 @@ public void concurrencyRegistrationTest() throws Exception { assertThat(c.size()).isEqualTo(1); } + @Test + public void testKafkaNullWithConcreteConsumerTypeNoLongerReachesFunctionAsRawMessage() { + // Regression test for #1448: a Kafka tombstone (KafkaNull payload) bound + // to a Consumer used to bypass conversion entirely and + // reach the function as a raw, unconverted Message, throwing an opaque + // "GenericMessage cannot be cast to " ClassCastException that gave + // no hint a Kafka tombstone was involved. + // + // With the fix, the message now flows through the same conversion path + // as any other message. For a plain (non-generic) declared parameter + // type such as this one, SmartCompositeMessageConverter's two-argument + // fromMessage(Message, Class) overload does not consult a registered + // MessageConverterHelper when every converter quietly returns null + // (only when a converter throws), so the failure still surfaces as a + // ClassCastException here rather than a MessageConversionException -- + // a pre-existing, orthogonal limitation of that overload, unrelated to + // this fix. But the payload is now unwrapped before the cast, so the + // exception names the real cause (KafkaNull) instead of the opaque + // wrapping Message type -- a genuine diagnostic improvement. + CompositeMessageConverter converter = new SmartCompositeMessageConverter( + List.of(new ByteArrayMessageConverter())); + + FunctionRegistration registration = new FunctionRegistration<>( + new ConsumePerson(), "consumePerson").type(ConsumePerson.class); + SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, converter, + new JacksonMapper(new ObjectMapper())); + catalog.register(registration); + FunctionInvocationWrapper lookedUpFunction = catalog.lookup("consumePerson"); + + Message kafkaNullMessage = MessageBuilder.withPayload((Object) KafkaNull.INSTANCE).build(); + + Assertions.assertThatThrownBy(() -> lookedUpFunction.apply(kafkaNullMessage)) + .isInstanceOf(ClassCastException.class) + .hasMessageContaining("KafkaNull") + .hasMessageNotContaining("GenericMessage"); + } + + @Test + public void testKafkaNullWithMessageTypedConsumerStillPassesThroughUnconverted() { + // A function genuinely declared to accept Message/KafkaNull must keep + // working exactly as before -- only the mismatched-type case changes. + ConsumeMessage function = new ConsumeMessage(); + FunctionRegistration registration = new FunctionRegistration<>( + function, "consumeMessage").type(ConsumeMessage.class); + SimpleFunctionRegistry catalog = new SimpleFunctionRegistry(this.conversionService, this.messageConverter, + new JacksonMapper(new ObjectMapper())); + catalog.register(registration); + FunctionInvocationWrapper lookedUpFunction = catalog.lookup("consumeMessage"); + + Message kafkaNullMessage = MessageBuilder.withPayload((Object) KafkaNull.INSTANCE).build(); + lookedUpFunction.apply(kafkaNullMessage); + + assertThat(function.received).isNotNull(); + assertThat(function.received.getPayload()).isSameAs(KafkaNull.INSTANCE); + } + @Test public void testCachingOfFunction() { Echo function = new Echo(); @@ -820,9 +877,24 @@ public Object apply(Object t) { } + private static final class ConsumePerson implements Consumer { + @Override + public void accept(Person person) { + fail("function must not be invoked when conversion fails"); + } + } + + private static final class ConsumeMessage implements Consumer> { + private volatile Message received; + + @Override + public void accept(Message message) { + this.received = message; + } + } + private static final class UpperCaseMessage implements Function, Message> { - @Override public Message apply(Message t) { return MessageBuilder.withPayload(t.getPayload().toUpperCase(Locale.ROOT)) diff --git a/spring-cloud-function-context/src/test/java/org/springframework/kafka/support/KafkaNull.java b/spring-cloud-function-context/src/test/java/org/springframework/kafka/support/KafkaNull.java new file mode 100644 index 000000000..9a6b66413 --- /dev/null +++ b/spring-cloud-function-context/src/test/java/org/springframework/kafka/support/KafkaNull.java @@ -0,0 +1,36 @@ +/* + * Copyright 2012-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.kafka.support; + +/** + * Minimal test double for {@code org.springframework.kafka.support.KafkaNull}. + * {@code SimpleFunctionRegistry} detects a Kafka tombstone payload by comparing + * {@code getClass().getName()} against this exact fully-qualified name (to avoid + * a hard compile dependency on spring-kafka), so a class with the same name and + * package is sufficient to exercise that code path in tests without pulling in + * the real spring-kafka dependency. + * + * @author Aditya Nikam + */ +public final class KafkaNull { + + public static final KafkaNull INSTANCE = new KafkaNull(); + + private KafkaNull() { + } + +}