KafkaNull payload bypasses MessageConverterHelper and is passed unconverted to a typed Consumer, causing ClassCastException

Describe the bug

When a Kafka consumer function is bound to a concrete argument type (e.g. Consumer<MyDomainType>), and the broker delivers a record with a null value — whether from log-compaction tombstone semantics, or simply a producer publishing an empty/null payload — Spring Cloud Stream's Kafka binder wraps it as a Message whose payload is KafkaNull.INSTANCE. SimpleFunctionRegistry.FunctionInvocationWrapper#convertInputIfNecessary detects this and returns the unconverted, raw Message as the function's input argument, instead of attempting conversion or signalling a conversion failure:

https://github.com/spring-cloud/spring-cloud-function/blob/main/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

else if (input instanceof Message) {
    input = this.filterOutHeaders((Message) input);
    if (((Message) input).getPayload().getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) {
        return input;   // bypasses every converter and the MessageConverterHelper hook entirely
    }
    ...

The generated bridge method for the functional interface then tries to cast that Message to the function's declared argument type, throwing a bare ClassCastException from SimpleFunctionRegistry.FunctionInvocationWrapper#invokeConsumer.

Critically, this happens without ever calling any registered MessageConverter or MessageConverterHelper, so the extension point added to resolve #1168 (letting an application signal "this should be a fatal conversion failure, not passed to my function") has no effect here — the KafkaNull branch returns early before that machinery is ever reached. Frameworks and applications built on top of this therefore see a plain ClassCastException with no way to distinguish "the framework couldn't convert this" from "a bug in my own consumer code."

This is closely related to #1168, which covers the case where a registered converter throws during conversion. This issue covers a narrower, still-open gap: the KafkaNull payload never reaches a converter at all, so #1168's fix doesn't help here.

Environment

• spring-cloud-function-context 4.3.4
• spring-cloud-stream / spring-cloud-stream-binder-kafka 4.3.3
• spring-kafka 3.3.16
• (the shortcut appears structurally unchanged across the versions we checked, so this likely affects a wide range)

To Reproduce

1. Define a functional Kafka consumer bound to a concrete type, e.g. Consumer<Event<SomeDomainType>>.
2. Publish a Kafka record with a null value to the bound topic.
3. Observe: the consumer function is invoked with a raw Message object instead of the declared type, throwing ClassCastException (e.g. class org.springframework.messaging.support.GenericMessage cannot be cast to class com.example.SomeDomainType).

We can provide a minimal standalone Spring Boot + Testcontainers-Kafka reproduction project if useful.

Expected behavior

When the payload is KafkaNull and the target function's declared argument type is not itself Message/KafkaNull-compatible, the framework should route through the same MessageConverterHelper.shouldFailIfCantConvert mechanism used elsewhere — either raising a MessageConversionException or letting the registered helper decide — rather than unconditionally passing the raw Message through. A function genuinely declared to accept Message<?>/KafkaNull (a real null-value-aware consumer) should keep working exactly as it does today; only the mismatched-type case should change.

Suggested fix

In the KafkaNull branch of convertInputIfNecessary, check whether the function's declared parameter type is assignable from Message/KafkaNull before returning the raw input. If it isn't, construct a MessageConversionException describing the failed conversion and call messageConverterHelper.shouldFailIfCantConvert(message, thatException) — passing the actual exception, not null (note: shouldFailIfCantConvert short-circuits to false whenever the throwable argument is null, both in the default implementation and in ours, so the framework needs to construct and pass a real exception here, not signal with a null throwable). If the helper returns true, throw the exception instead of returning the raw message. This reuses the exact mechanism #1168 introduced rather than adding a new one.

Current workaround

We've worked around this downstream, in our own Kafka error-handling layer, by pattern-matching the resulting ClassCastException's message (checking that the class being cast from is GenericMessage) and reclassifying it as a MessageConversionException after the fact. It works, but it's inherently fragile (string-matching an exception message), and only possible because we own that error-handling layer — most Spring Cloud Stream users wouldn't have that option.

Comienza en spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java, en FunctionInvocationWrapper#convertInputIfNecessary y el retorno temprano de KafkaNull. Compara esa ruta con el flujo de MessageConverterHelper introducido para #1168 y reprodúcela con un consumidor de Kafka tipado que reciba un registro nulo. Se considera terminado cuando las cargas KafkaNull incompatibles producen el error de conversión previsto, mientras que los consumidores compatibles con Message/KafkaNull conservan el comportamiento actual.