From 35c8238d3e57a646203a2c0cc21528d860385835 Mon Sep 17 00:00:00 2001 From: adityaanikam Date: Wed, 2 Sep 2026 17:09:12 +0530 Subject: [PATCH] fix(context): route KafkaNull payloads through conversion for non-Message types convertInputIfNecessary returned the raw, unconverted Message as soon as it saw a KafkaNull payload, regardless of what the target function declared as its input type. For a function bound to a concrete type such as Consumer, that Message was then cast to the declared type, throwing "class GenericMessage cannot be cast to class MyType" -- an error that gives no indication a Kafka tombstone was involved. The shortcut also ran before any MessageConverter or MessageConverterHelper was consulted, so the hook added in gh-1168 could not see these messages at all. Guard the shortcut with isInputTypeMessage(), the same predicate the class already uses elsewhere to test whether the declared input type is itself Message-compatible. A function genuinely declared to accept Message still receives the raw message exactly as before; anything else now follows the normal conversion path. Fixes gh-1448 Signed-off-by: adityaanikam --- .../catalog/SimpleFunctionRegistry.java | 3 +- .../catalog/SimpleFunctionRegistryTests.java | 74 ++++++++++++++++++- .../kafka/support/KafkaNull.java | 36 +++++++++ 3 files changed, 111 insertions(+), 2 deletions(-) create mode 100644 spring-cloud-function-context/src/test/java/org/springframework/kafka/support/KafkaNull.java 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() { + } + +}