消费泛型类型事件负载时出现java.lang.ClassCastException异常

泛型类型擦除导致反序列化失败。Spring Cloud Stream处理泛型消息时，无法从擦除后的PayloadWithEmbeddedHeaders<T>推断出T的具体类型，实际Kafka消费时会将body反序列化为LinkedHashMap而非目标InstructionalPlanCreatedEvent对象。测试环境因直接传递强类型对象未触发问题，但JSON反序列化时泛型信息丢失引发ClassCastException。

方案1：创建泛型的具体子类（推荐） #

定义明确的非泛型子类，消除类型擦除影响，让反序列化器直接识别目标类型：

// 自定义具体子类，明确泛型参数类型
data class InstructionalPlanCreatedPayload(
    override val headers: Map<String, Any> = emptyMap(),
    override val body: InstructionalPlanCreatedEvent
) : PayloadWithEmbeddedHeaders<InstructionalPlanCreatedEvent>(headers, body)

修改消费者函数的泛型声明：

@Bean
fun ipConsumer(): Consumer<Message<InstructionalPlanCreatedPayload>> {
    return Consumer { message ->
        if (message.headers[PublisherKafka.EVENT_TYPE_HEADER_NAME] == InstructionalPlanCreatedEvent::class.java.simpleName) {
            process(ip = message.payload.body)
        }
    }
}

方案2：手动转换LinkedHashMap为目标对象 #

借助Jackson的ObjectMapper将反序列化后的LinkedHashMap强制转换为事件类，适合无法修改泛型结构的场景：

@Bean
fun ipConsumer(objectMapper: ObjectMapper): Consumer<Message<PayloadWithEmbeddedHeaders<*>>> {
    return Consumer { message ->
        val eventType = message.headers[PublisherKafka.EVENT_TYPE_HEADER_NAME]?.toString()
        if (eventType == InstructionalPlanCreatedEvent::class.java.simpleName) {
            val event = objectMapper.convertValue(message.payload.body, InstructionalPlanCreatedEvent::class.java)
            process(ip = event)
        }
    }
}

方案3：配置Kafka序列化/反序列化器传递类型信息 #

通过配置绑定器，让生产者发送类型元数据，消费者依据元数据完成正确反序列化：

生产者配置（application.properties） #

spring.cloud.stream.kafka.binder.configuration.value.serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.cloud.stream.kafka.binder.configuration.json.type.mapping=InstructionalPlanCreatedEvent:com.yourpackage.InstructionalPlanCreatedEvent

消费者配置（application.properties） #

spring.cloud.stream.kafka.binder.configuration.value.deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.cloud.stream.kafka.binder.configuration.spring.json.trusted.packages=com.yourpackage
spring.cloud.stream.kafka.binder.configuration.json.type.mapping=InstructionalPlanCreatedEvent:com.yourpackage.InstructionalPlanCreatedEvent

消费者函数可保留原有泛型声明：

@Bean
fun ipConsumer(): Consumer<Message<PayloadWithEmbeddedHeaders<InstructionalPlanCreatedEvent>>> {
    return Consumer { message ->
        if (message.headers[PublisherKafka.EVENT_TYPE_HEADER_NAME] == InstructionalPlanCreatedEvent::class.java.simpleName) {
            process(ip = message.payload.body)
        }
    }
}

内容的提问来源于stack exchange，提问作者Faizan