能否查看嵌入式Kafka Topic中的内容？IntelliJ调试无果

嵌入式Kafka（比如Spring Kafka Test中的EmbeddedKafka）没有直接在IntelliJ调试GUI中查看Topic内容的内置选项，但可以通过以下方式实现类似效果：

• 调试时执行临时消费代码
  利用IntelliJ的「Evaluate Expression」功能（快捷键Alt+F8），在断点处执行一段消费Topic的代码，直接获取并打印内容：

  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.KafkaConsumer;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.common.serialization.StringDeserializer;
  import java.util.Collections;
  import java.util.Properties;
  import java.time.Duration;
  
  // 替换成你的嵌入式Kafka Broker地址和Topic名
  String bootstrapServers = "localhost:9092"; // 或者从embeddedKafka实例获取：embeddedKafka.getBrokersAsString()
  String topic = "your-target-topic";
  
  Properties props = new Properties();
  props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
  props.put(ConsumerConfig.GROUP_ID_CONFIG, "debug-consumer-group");
  props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
  props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  
  try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
      consumer.subscribe(Collections.singletonList(topic));
      ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
      records.forEach(record -> System.out.printf("偏移量: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value()));
  }
  

  执行后，控制台会输出该Topic的所有消息。

• 提前编写调试工具方法
  在测试代码中定义一个工具方法，调试时直接调用即可查看指定Topic内容：

  public static void inspectEmbeddedKafkaTopic(String bootstrapServers, String topicName) {
      Properties props = new Properties();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
      props.put(ConsumerConfig.GROUP_ID_CONFIG, "debug-inspect-group");
      props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
  
      try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
          consumer.subscribe(Collections.singletonList(topicName));
          ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(2));
          System.out.println("=== Topic [" + topicName + "] 内容 ===");
          records.forEach(r -> System.out.println(String.format("偏移量: %d | Key: %s | Value: %s", r.offset(), r.key(), r.value())));
          System.out.println("=== 内容结束 ===");
      }
  }
  

  在断点处调用inspectEmbeddedKafkaTopic(embeddedKafka.getBrokersAsString(), "your-topic")，就能快速打印Topic内容。

• 注意事项

  • 确保调试时的消费者组ID是临时的，避免影响其他测试的偏移量
  • 如果消息是自定义序列化的，要替换对应的反序列化类（比如用JsonDeserializer替代StringDeserializer）

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