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

Offen
#1,448 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Dieses Issue hat noch niemand übernommen.

Bewertung

Schwierigkeit
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
55/100
Issue-Typ
Bug
Klarheit
Klar beschrieben
Aktivitätsstatus
Aktiv
Tech-Stack
java, kafka, spring
Bereich
backend

Rechercherichtung

Beginne in spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java bei FunctionInvocationWrapper#convertInputIfNecessary und der frühen Rückgabe für KafkaNull. Vergleiche diesen Pfad mit dem für #1168 eingeführten MessageConverterHelper-Ablauf und reproduziere ihn mit einem typisierten Kafka-Consumer, der einen Null-Record empfängt. Erledigt ist die Aufgabe, wenn inkompatible KafkaNull-Payloads den vorgesehenen Konvertierungsfehler erzeugen, während mit Message/KafkaNull kompatible Consumer ihr bisheriges Verhalten beibehalten.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

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.

Vorherrschende Sprache
Java
Sterne
1.1k
Forks
641
Ø Merge
11 Std. 2 Min.
Gemergte PRs (30 T.)
8

Beitragsleitfaden

Beitragsleitfaden öffnen

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus spring-cloud/spring-cloud-function

Alle Issues in spring-cloud/spring-cloud-function

Ähnliche Issues

Weitere Issues zu Java

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.