Spring Kafka Binder已连接至主题但无法接收消息问题排查求助

从你的日志和配置来看，应用已经成功加入消费者组、分配了TOPIC-0分区，那收不到消息大概率不是连接或组管理的问题，咱们从以下几个方向逐一排查：

1. 先确认偏移量是否正确 #

日志里显示你的消费者当前提交的偏移量是1076：

Setting offset for partition TOPIC-0 to the committed offset FetchPosition{offset=1076...}

这时候要确认：

• 生产者发送的消息是不是在这个偏移量之后？如果生产者发的消息都在1076之前，那消费者已经消费过（或者提交过这个偏移），自然不会再收到。
• 你可以用Kafka的命令行工具查看消费组的偏移情况：

kafka-consumer-groups.sh --describe --group service-dev --bootstrap-server port1.test.com:6667

对比CURRENT-OFFSET和LOG-END-OFFSET，如果两者相等，说明已经追上了最新消息，没有未消费的内容。

2. 序列化/反序列化不匹配 #

你的监听器方法参数是Object msg，Spring Cloud Stream默认用JSON序列化器。如果生产者用的是其他序列化方式（比如StringSerializer、自定义序列化器），消费者可能反序列化失败，消息被静默丢弃（默认配置下不会抛出明显错误）。

可以做这些测试：

• 临时把监听器参数改成byte[]或String，直接接收原始消息：

@StreamListener(target = "stream-input")
public void processMessage(byte[] msg) {
    log.info("Received raw message: {}", new String(msg));
}

如果能收到内容，说明是序列化配置的问题，需要调整消费者的content-type配置，比如生产者发的是字符串，就加：

spring.cloud.stream.bindings.stream-input.content-type=text/plain

• 开启DEBUG日志，查看反序列化相关的日志：把org.springframework.cloud.stream和org.apache.kafka.clients.consumer的日志级别设为DEBUG，看有没有反序列化异常。

3. 生产者端的配置匹配问题 #

（1）Topic名称大小写匹配 #

Kafka的Topic是大小写敏感的，确认生产者发送的Topic名称和你配置的TOPIC完全一致（包括大小写）。

（2）安全配置对齐 #

你的消费者用了Kerberos认证（SASL_PLAINTEXT），那生产者也必须配置对应的安全参数，否则根本无法向这个Topic发送消息。同时要检查Kafka的ACL权限：

• 确认你的消费者账号kafka_user@TEST.COM有TOPIC的消费权限
• 确认生产者的账号有TOPIC的写入权限

（3）分区策略匹配 #

如果Topic有多个分区，但你的消费者组只有一个实例，只会分配一个分区（比如你日志里的TOPIC-0）。如果生产者的消息都发送到了其他分区，那这个消费者自然收不到。可以用命令行查看Topic的分区数：

kafka-topics.sh --describe --topic TOPIC --bootstrap-server port1.test.com:6667

4. 关于日志里的「分区撤销」 #

你日志里的Revoking previously assigned partitions []是正常流程，因为第一次加入组时，消费者之前没有分配过任何分区，所以撤销的是空集合，和收不到消息没有关系，不用在意这个。

5. 快速验证：用命令行消费者测试 #

最直接的方式是用Kafka控制台消费者，用和你应用相同的配置来消费TOPIC，看看能不能收到消息：

1. 创建一个consumer.properties文件：

security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka
sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="C:\\Users\\src\\main\\resources\\kafka\\kafka_user.keytab" principal="kafka_user@TEST.COM";

2. 启动控制台消费者：

kafka-console-consumer.sh --bootstrap-server port1.test.com:6667 --topic TOPIC --group service-dev --consumer-config consumer.properties

• 如果控制台能收到消息：问题出在你的应用代码或配置上，回到前面的序列化、Spring Cloud Stream配置排查
• 如果控制台也收不到：问题出在生产者或Kafka集群的权限/Topic配置上，需要检查生产者的发送逻辑和Kafka的ACL

6. Spring Cloud Stream版本与配置细节 #

如果你用的是Spring Cloud Stream 3.x及以上版本，@EnableBinding已经被标记为Deprecated，虽然旧代码仍能运行，但可能存在一些兼容性问题。如果你的依赖版本是较新的，建议尝试用函数式编程的方式定义消费者：

@Bean
public Consumer<Message<Object>> streamInput() {
    return msg -> {
        log.info("Received message: {}", msg.getPayload());
        // 处理逻辑
    };
}

同时调整yaml配置，去掉@EnableBinding相关的内容，简化绑定配置：

spring.cloud.stream.bindings.streamInput-in-0.destination=TOPIC
spring.cloud.stream.bindings.streamInput-in-0.group=service-dev

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