Spring Cloud Stream批量消费Kafka Avro消息时Payload为空的解决方法

1. 修复版本兼容性问题 #

移除spring-cloud-stream-schema 2.2.1.RELEASE依赖，该版本与Spring Cloud Stream 4.x（基于Spring Boot 3.x）存在兼容性冲突。改用Confluent官方的Avro序列化依赖：

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
    <version>7.4.0</version> <!-- 版本需匹配Spring Cloud Stream 4.0.2对应的Kafka 3.4.x版本 -->
</dependency>

2. 正确配置批量消费与手动ACK #

在application.yml中添加以下配置：

spring:
  cloud:
    stream:
      bindings:
        input: # 你的消费者绑定名称
          destination: your-avro-topic # 目标Avro主题名
          group: your-consumer-group # 必须指定消费组
          consumer:
            batch-mode: true # 开启批量消费模式
      kafka:
        bindings:
          input:
            consumer:
              auto-commit-offset: false # 关闭自动提交，启用手动ACK
              max-poll-records: 10 # 批量拉取的消息数量，按需调整
              configuration:
                schema.registry.url: http://your-schema-registry:8081 # Schema Registry地址
                value.deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer
                key.deserializer: org.apache.kafka.common.serialization.StringDeserializer # 按实际key类型调整
                specific.avro.reader: true # 启用具体Avro类反序列化，必须设置

3. 编写正确的批量消费代码 #

使用函数式编程模型，接收List<Message<MyAvroObject>>类型输入，从每个Message中获取Payload并完成手动ACK：

import org.springframework.messaging.Message;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.List;
import java.util.function.Consumer;

@Configuration
public class BatchConsumerConfig {

    @Bean
    public Consumer<List<Message<MyAvroObject>>> batchAvroConsumer() {
        return messages -> {
            for (Message<MyAvroObject> message : messages) {
                MyAvroObject payload = message.getPayload();
                // 执行业务逻辑处理
                System.out.println("处理批量消息：" + payload);

                // 手动ACK当前消息
                Acknowledgment ack = message.getHeaders().get(KafkaHeaders.ACKNOWLEDGMENT, Acknowledgment.class);
                if (ack != null) {
                    ack.acknowledge();
                }
            }
        };
    }
}

关键注意事项 #

• 消息类型不要混淆：批量模式下，函数输入必须是List<Message<T>>，而非Message<List<T>>，后者会导致Payload为空。
• specific.avro.reader必须启用：确保反序列化时使用生成的具体Avro类（如MyAvroObject），而非通用的GenericRecord。
• 批量ACK可选：若无需单条ACK，可从批次中任意一个Message的Header获取Acknowledgment，调用一次acknowledge()即可提交整个批次的偏移量。

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