Spring Cloud Stream Kafka消费者自定义消息反序列化问题

问题背景 #

使用Spring Cloud Stream对接Kafka时，消费者端无法自动将JSON消息反序列化为目标DTO，只能手动处理，寻求优雅配置方案。

相关配置与代码 #

Spring配置（application.yaml） #

cloud:
  stream:
    bindings:
      processData-in-0:
        destination: test.public.huge_test_table
        group: my-consumer-group
        content-type: application/json
        consumer:
          concurrency: 10
    kafka:
      binder:
        brokers: localhost:9092

Kafka消息Payload格式 #

{
    "metadata": {
         // 省略其他字段
    },
    "payload": {
        "title": "Whatever"
        // 省略其他字段
    },
    "op": "c"
}

DTO定义（Kotlin） #

data class Foo<T> (
    @JsonProperty("payload")
    val payload: T,
    @JsonProperty("op")
    val op: String
)

data class Bar (
    @JsonProperty("title")
    val title: String,
    // 省略其他字段
)

消费者代码尝试 #

@Configuration
class MyListener {

    private val logger = KotlinLogging.logger {}

    @Bean
    fun processData(): Consumer<Message<Foo<Bar>>> {
        return Consumer { message ->
            logger.info { "Received message: ${message.payload}" }
        }
    }
}

错误信息 #

运行时抛出类型转换异常：

class [B cannot be cast to class com.example.dto.FooDTO ([B is in module java.base of loader 'bootstrap'; com.example.dto.FooDTO is in unnamed module of loader 'app')

发现GenericMessage的payload实际为ByteArray，改用Consumer<Message<String>>可正常获取JSON字符串。

已尝试的方案 #

1. 手动反序列化（可行，但不够优雅）：

@Bean
fun processData(): Consumer<Message<String>> {
    return Consumer { message ->
        val payload = objectMapper.readValue<Foo<Bar>>(message.payload.toString())
        logger.info { "Received message: ${payload}" }
    }
}

2. 自定义MessageConverter Bean：

@Bean
fun customMessageConverter(objectMapper: ObjectMapper): MessageConverter {
    val converter = MappingJackson2MessageConverter()
    converter.objectMapper = objectMapper
    return converter
}

补充疑问 #

进一步排查发现反序列化失败是因为JSON包含DTO未定义的字段，已设置spring.jackson.deserialization.fail-on-unknown-properties=false（默认值），但只有给DTO添加@JsonIgnoreProperties(ignoreUnknown = true)注解才生效，想了解原因。

解决方案 #

1. 实现自动反序列化的正确配置 #

要让Spring Cloud Stream自动完成JSON到DTO的反序列化，需满足以下配置要求：

• 强制禁用原生解码：在消费者绑定配置中添加use-native-decoding: false，强制使用Spring的消息转换器而非Kafka原生序列化器，确保content-type: application/json配置生效：

  processData-in-0:
    # 其他配置...
    consumer:
      concurrency: 10
      use-native-decoding: false
  

• 直接消费DTO类型：无需用Message<Foo<Bar>>包装，直接消费Foo<Bar>即可，Spring Cloud Stream会自动处理消息转换：

  @Bean
  fun processData(): Consumer<Foo<Bar>> {
      return Consumer { payload ->
          logger.info { "Received message: $payload" }
      }
  }
  

  若需要获取消息头，可继续使用Message<Foo<Bar>>，但必须确保上述配置已正确生效。

2. 关于未知字段忽略的问题 #

spring.jackson.deserialization.fail-on-unknown-properties=false全局配置不生效的核心原因：
Spring Cloud Stream的Kafka binder默认使用Kafka原生Jackson反序列化器，而非Spring MVC/Web中使用的MappingJackson2HttpMessageConverter。原生反序列化器不会读取Spring的全局Jackson配置，仅识别DTO类上的@JsonIgnoreProperties注解。

若要让全局配置生效，需自定义Kafka反序列化器并绑定Spring上下文的ObjectMapper：

@Bean
fun consumerConfigurer(): ConsumerConfigurer {
    return ConsumerConfigurer { props, _ ->
        props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = JsonDeserializer::class.java
        props[JsonDeserializer.VALUE_DEFAULT_TYPE] = Foo::class.java.name
        props[JsonDeserializer.USE_TYPE_INFO_HEADERS] = false.toString()
        props[JsonDeserializer.OBJECT_MAPPER_BEAN_NAME] = "objectMapper"
    }
}

@Bean
fun objectMapper(): ObjectMapper {
    return ObjectMapper()
        .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
        .registerModule(KotlinModule())
}

内容的提问来源于stack exchange，提问作者João Menighin