Using generic container as @KafkaHandler parameter?

[Gyllsdorff]

This might not be doable due to JVM:s generic type erasure, but I just want to check in case I am missing something obvious. :)

I have a kafka topic where the payload have a set of standard fields and a data object that can have 3 different shapes. The kafka message contains a header where the value is the type of the object in the data field.

I can solve this by having separate concrete Container types for each data object type but I would like to use a generic container since the root object have a lot of fields in our production classes.

The json deserialization works fine but the DelegatingInvocableHandler.findHandlerForPayload function seems to look for a handler with a Container parameter without the type so I get a No method found for class com.example.Container error or Cannot convert from [com.example.Container] to [com.example.Container] from the InvocableHandlerMethod handler if I use @KafkaListener directly on a function.

Kotlin code example

data class Container<T>(val id: String, val data: T)
data class Default(val default: String?)
data class Foo(val value: String)
data class Bar(val count: Int)

val defaultSchema = TypeFactory.defaultInstance().constructParametricType(Container::class.java, Default::class.java)
val schemaNameToType: Map<String, JavaType> = mapOf(
    "foo" to Foo::class.java,
    "bar" to Bar::class.java,
)
    .mapValues { TypeFactory.defaultInstance().constructParametricType(Container::class.java, it.value) }

fun kafkaMessageToClassType(
  data: ByteArray,
  headers: Headers?,
): JavaType {
    val schema = headers?.lastHeader("schema")?.value()?.toString(Charsets.UTF_8)
    if (schema == null) {
        return defaultSchema
    }
    return schemaNameToType[schema] ?: defaultSchema
}

@KafkaListener(topics = ["example"])
@Service
class Handler {
    @KafkaHandler
    fun processFoo(message: Container<Foo>) {
        println("Foo")
    }

    @KafkaHandler
    fun processBar(message: Container<Bar>) {
        println("Bar")
    }
}

[Gyllsdorff]

@Service
class Handler {
    @KafkaListener(topics = ["example"])
    fun process(message: ConsumerRecord<String, Container<*>>) {
        when (val data = message.value().data) {
            is Foo -> logger.info { "Foo $data" }
            is Bar -> logger.info { "Bar $data" }
            else -> logger.warn { "Unknown data type: $data" }
      }
    }
}

This is the best and least-hacky solution I have found so far but I would like to remove the dependency on the Spring Kafka ConsumerRecord class.