Kafka消费者无输出结果？自研Java消费程序异常排查求助

看起来你已经通过官方的kafka-console-consumer.sh确认了Topic里有完整数据，那问题大概率出在你的DumpKafkaTopic消费程序本身，我给你几个常见的排查方向和修复建议：

1. 检查消息反序列化配置是否匹配 #

这是最容易踩的坑！如果你的生产者TsdbMetricToKafkaTopic用了特定的序列化器（比如自定义的OpenTSDB指标序列化、或者特定的JSON序列化），而消费程序用了默认的StringDeserializer，就会导致消息解析不全，出现截断或乱码（比如你看到的res...）。

• 先确认生产者的序列化配置：比如生产者端是不是用了org.apache.kafka.common.serialization.StringSerializer，还是自定义的序列化类？
• 消费程序必须用完全匹配的反序列化器，比如生产者用了自定义的TsdbMetricSerializer，消费端就要对应配置TsdbMetricDeserializer。

举个String类型消息的配置示例：

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-broker:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "dump-topic-group");
// 重点：确保key和value的反序列化器和生产者完全一致
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

2. 检查消费程序的消息处理逻辑 #

可能你的程序在输出消息时做了无意识的截断操作：

• 是不是写了System.out.println(message.substring(0, 3))这类截断代码？
• 或者自定义的OpenTSDB指标对象，重写的toString()方法只输出了部分字段（比如只打印了res开头的内容）？

建议直接打印原始的完整消息内容：

ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    // 直接打印完整的value，不要做任何截断
    System.out.println("完整消息内容: " + record.value());
    // 如果是自定义对象，直接打印对象本身或者调用完整的toString实现
    // System.out.println("完整指标对象: " + record.value().toString());
}

3. 确认消费偏移量的起始位置 #

有时候消费程序可能默认从最新的偏移量开始消费，导致只拿到了部分新消息，而不是Topic里的全部历史数据。可以在消费配置里强制从头开始消费（适合调试场景）：

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

另外，也可以用Kafka官方工具查看消费组的偏移量状态，确认是否有未消费的消息：

kafka-consumer-groups.sh --bootstrap-server your-kafka-broker:9092 --describe --group dump-topic-group

4. 检查消息拉取的超时时间与循环逻辑 #

如果你的poll()超时时间设置得太短，可能还没拉取完所有消息，程序就提前结束了。调试时可以把超时时间设长一点，或者用循环持续拉取直到没有新消息：

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
    if (records.isEmpty()) {
        // 没有新消息时退出循环（仅调试用）
        break;
    }
    for (ConsumerRecord<String, String> record : records) {
        System.out.println("消息内容: " + record.value());
    }
}

5. 检查消息大小限制配置 #

虽然你用console consumer能看到完整数据，但也要确认消费程序有没有设置过小的消息拉取限制：

• 如果生产者发送的消息超过了消费端默认的max.partition.fetch.bytes（默认1MB），可能会导致消息被截断。
• 可以在消费配置里调大这个参数：

props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 10*1024*1024); // 设置为10MB

先从这几个方向排查，应该能快速定位到问题！

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