使用Kafka JDBC Sink Connector时，无需kcat查看死信队列失败记录头部的方法咨询

没问题，不用kcat也能查看死信队列的头部信息，下面给你两种可行的方法，同时也帮你排查Reporter主题未创建的问题：

查看死信队列头部信息（无需kcat） #

方法1：用Kafka自带的控制台消费者工具（Windows环境） #

Kafka的kafka-console-consumer.bat支持通过配置属性打印消息头部，只需要添加几个参数就能显示死信队列里的错误相关头部（比如kafka_connect_error_message、kafka_connect_error_stacktrace这些Connect自动注入的关键信息）。

执行以下命令：

kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic failed_records --from-beginning ^
--property print.headers=true ^
--property print.key=true ^
--property print.value=true ^
--property header.separator=" | " ^
--property headers.format="%k=%s"

参数说明：

• print.headers=true：开启头部打印
• header.separator：设置头部之间的分隔符，让输出更易读
• headers.format：定义头部的显示格式，这里是键=值的形式

执行后你就能在输出里看到包含错误原因、堆栈信息的头部内容了。

方法2：编写简单的Java消费者程序 #

如果控制台工具的输出不够灵活，你可以自己写一段极简的Java代码来消费并解析头部，步骤如下：

1. 确保项目引入Kafka Java客户端依赖（比如Maven依赖org.apache.kafka:kafka-clients:2.8.0，版本和你的Kafka集群匹配）
2. 编写消费代码：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class DLQHeaderViewer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "dlq-header-viewer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        try (Consumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("failed_records"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                records.forEach(record -> {
                    System.out.println("---------- 消息内容 ----------");
                    System.out.println("Key: " + record.key());
                    System.out.println("Value: " + record.value());
                    System.out.println("---------- 头部信息 ----------");
                    for (Header header : record.headers()) {
                        String headerValue = new String(header.value());
                        System.out.printf("Header [%s]: %s%n", header.key(), headerValue);
                    }
                });
            }
        }
    }
}

运行这段代码后，就能清晰看到每条死信消息的所有头部字段，包括Connect自动添加的错误详情。

解决Connect Reporter主题未创建的问题 #

你的success-responses和error-responses主题没生成，大概率是以下几个原因，按顺序排查：

1. 检查Kafka自动创建主题开关：确认Kafka broker的server.properties里auto.create.topics.enable=true（默认是开启的，如果手动改过需要改回来），Reporter依赖Kafka自动创建主题。
2. 验证Reporter配置正确性：确认reporter.bootstrap.servers和你的Kafka集群地址一致（你这里是localhost:9092，如果是Windows下本地集群应该没问题，但要确保Connect能访问到这个地址）。
3. 查看Connect日志：去Kafka的logs目录下找connect.log或connect-distributed.log，搜索reporter相关的错误信息，比如权限不足、连接失败等。
4. 手动创建主题：如果自动创建失效，直接用Kafka的主题管理工具手动创建：

kafka-topics.bat --create --bootstrap-server localhost:9092 --topic success-responses --replication-factor 1 --partitions 1
kafka-topics.bat --create --bootstrap-server localhost:9092 --topic error-responses --replication-factor 1 --partitions 1

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