fix(context): route KafkaNull payloads through conversion for non-Message types

[adityaanikam]

Fixes gh-1448

Problem

FunctionInvocationWrapper#convertInputIfNecessary returns the raw, unconverted Message as soon as it sees a KafkaNull payload:

else if (input instanceof Message) {
    input = this.filterOutHeaders((Message) input);
    if (((Message) input).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) {
        return input;
    }

That shortcut runs regardless of what the target function declared as its input type. For a function bound to a concrete type (Consumer<MyType>), the returned Message is then cast to the declared type at invocation, producing:

class org.springframework.messaging.support.GenericMessage cannot be cast to class com.example.MyType

which gives no indication that a Kafka tombstone was involved. The shortcut also returns before functionInvocationHelper.preProcessInput and convertInputMessageIfNecessary are reached, so no MessageConverter — and therefore no MessageConverterHelper — ever sees these messages. The hook added for gh-1168 cannot apply here.

Fix

Guard the shortcut with isInputTypeMessage(), the predicate this class already uses a few dozen lines earlier for the same question:

if (this.isInputTypeMessage()
        && ((Message) input).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) {
    return input;
}

A function genuinely declared to accept Message<?> still receives the raw message exactly as it does today. Anything else now follows the normal conversion path instead of bypassing it.

I did not implement the alternative suggested on the issue (constructing a MessageConversionException and calling shouldFailIfCantConvert directly from SimpleFunctionRegistry), because messageConverterHelpersSupplier and failConversionIfNecessary are both private to SmartCompositeMessageConverter and SimpleFunctionRegistry holds its converter as a CompositeMessageConverter. Doing it that way would need new public API; letting the message follow the existing path does not.

Scope of the improvement

Worth being precise about what this does and does not change downstream, since it depends on the shape of the declared type:

• For a parameterized declared type, convertInputMessageIfNecessary calls fromMessage(Message, Class, Object), which calls failConversionIfNecessary when no converter succeeds — so a registered MessageConverterHelper is consulted and can raise MessageConversionException.
• For a plain (non-generic) declared type such as Consumer<MyType>, it calls the two-argument fromMessage(Message, Class), which only calls failConversionIfNecessary when a converter throws, not when converters quietly return null. So that path still ends in a ClassCastException.

In the second case the improvement is narrower but real: the payload is now unwrapped before the cast, so the error names the actual cause:

class org.springframework.kafka.support.KafkaNull cannot be cast to class com.example.MyType

instead of pointing at the wrapping GenericMessage. The asymmetry between the two fromMessage overloads' failure semantics looks orthogonal to this issue, so I left it alone rather than widen the change.

Testing

Two tests in SimpleFunctionRegistryTests:

• testKafkaNullWithConcreteConsumerTypeNoLongerReachesFunctionAsRawMessage — a Consumer<Person> receiving a KafkaNull payload; asserts the failure names KafkaNull and not GenericMessage.
• testKafkaNullWithMessageTypedConsumerStillPassesThroughUnconverted — a Consumer<Message<Object>> still receives the raw message with the KafkaNull payload intact.

SimpleFunctionRegistry detects the tombstone by comparing getClass().getName() against the literal class name rather than by type, so the tests use a same-named test class under src/test/java/org/springframework/kafka/support/ instead of adding a spring-kafka dependency.

Verified with a negative control: reverting only the source change (keeping both tests) makes the first test fail with class org.springframework.messaging.support.GenericMessage cannot be cast to class ...Person, matching the symptom reported on the issue; restoring it brings the suite back to 34 passing, 2 pre-existing skips.