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<MyType>, 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
Resolves #1456

Signed-off-by: adityaanikam <adityanikam9502@gmail.com>

Author: adityaanikam

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;
 				}
 

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<ConcreteType> used to bypass conversion entirely and
+		// reach the function as a raw, unconverted Message, throwing an opaque
+		// "GenericMessage cannot be cast to <Type>" 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<ConsumePerson> 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<Object> 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<ConsumeMessage> 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<Object> 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<Person> {
+		@Override
+		public void accept(Person person) {
+			fail("function must not be invoked when conversion fails");
+		}
+	}
+
+	private static final class ConsumeMessage implements Consumer<Message<Object>> {
+		private volatile Message<Object> received;
+
+		@Override
+		public void accept(Message<Object> message) {
+			this.received = message;
+		}
+	}
+
 	private static final class UpperCaseMessage
 			implements Function<Message<String>, Message<String>> {
-
 		@Override
 		public Message<String> apply(Message<String> t) {
 			return MessageBuilder.withPayload(t.getPayload().toUpperCase(Locale.ROOT))

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() {
+	}
+
+}