Spring Cloud Stream Kafka Binder序列化器配置未生效问题

生产者自定义参数不生效的核心原因是Spring Cloud Stream配置层级写错：

• Kafka binder的专属生产者配置，不能直接放在spring.cloud.stream.bindings.<binding-name>.producer.configuration路径下，该路径是通用binder的配置占位，Kafka相关的自定义参数必须放到spring.cloud.stream.kafka.bindings.<binding-name>.producer.configuration节点下才会被Kafka binder加载。
• 你还将Kafka原生的点分隔参数（如max.block.ms）错误拆成了多层YAML嵌套节点，导致参数无法正确映射。
  最终框架直接加载了默认生产者配置：key/value序列化器使用ByteArraySerializer，acks、重试次数、幂等性等参数都走默认值，和你预期的配置完全不符，因此String类型的消息key无法被序列化，抛出类型转换异常。
  另外你定义的producer函数是无业务逻辑的空透传实现，完全可以删除，Stream Bridge本身支持动态发送消息，不需要提前定义空生产函数做转发，多余的函数定义只会增加不必要的绑定初始化开销。

1. 修正application.yaml配置结构 #

将Kafka相关配置移到正确的节点下，参数名保持和Kafka原生配置一致，修正后的配置如下：

spring:
  main:
    banner-mode: off
  mongodb:
    embedded:
      version: 3.4.6
  data:
    mongodb:
      port: 29129
      host: localhost
      database: howler_db
  kafka:
    binder:
      brokers: localhost:9952
  cloud:
    stream:
      bindings:
        producer-out-0:
          # 通用绑定配置放在该节点下
          useTopicHeader: true
          content-type: application/json
      # Kafka binder专属配置统一放在kafka节点下
      kafka:
        bindings:
          producer-out-0:
            producer:
              # Kafka原生参数统一放在configuration下，不要拆多层嵌套
              configuration:
                key.serializer: org.apache.kafka.common.serialization.StringSerializer
                value.serializer: org.springframework.kafka.support.serializer.JsonSerializer
                retries: 3
                max.block.ms: 500
                enable.idempotence: true
                acks: all
              # Kafka重试主题专属配置放在producer层，不要塞入configuration
              retry-topic:
                delay: 200

注意：删除spring.cloud.function.definition下的producer配置，后续删掉空函数后不需要保留该定义。

2. 删除多余的空生产函数 #

移除代码中无实际逻辑的producerBean定义，streamBridge.send("producer", message)的调用逻辑不需要修改，框架会自动初始化对应动态binding。

// 删掉以下这段空实现即可
/*
@Bean
public Function<Flux<Message<?>>, Flux<Message<?>>> producer() {
    return mapFlux -> mapFlux.map(m -> MessageBuilder.withPayload(m.getPayload()).copyHeaders(m.getHeaders()).build());
}
*/

3. 验证配置生效 #

重启应用后查看启动日志中的ProducerConfig输出，确认以下参数和自定义配置一致：

• key.serializer = class org.apache.kafka.common.serialization.StringSerializer
• value.serializer = class org.springframework.kafka.support.serializer.JsonSerializer
• acks = all
• enable.idempotence = true
• retries = 3
  配置加载正确后，String类型的消息key可正常序列化，不会再抛出序列化异常。

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