如何实现Kafka topic消费者访问行为的日志记录与格式输出

方案一：消费者端埋点（成本最低、适配性最高） #

无需改动Kafka服务端配置，直接在消费者业务逻辑中增加日志输出即可：

• 代码中提前读取部署主机的IP（可按需求取内网IP或公网IP），在消费者拉取消息的逻辑后新增日志输出节点
• 按需求的格式拼接时间、主机IP、topic名称、对应分片/偏移量字段即可
• Java版示例代码片段：

// 提前读取本机IP
String localHostIp = InetAddress.getLocalHost().getHostAddress();
// 定义时间格式化规则
SimpleDateFormat dateFormat = new SimpleDateFormat("MM-dd-yyyy");
// 消费逻辑
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
    String logContent = String.format("%s IP:%s accessed topic \"%s\" fragment %d",
        dateFormat.format(new Date()), localHostIp, record.topic(), record.partition());
    // 可直接输出到本地日志文件，也可上报到统一日志收集系统
    log.info(logContent);
}

• 如果日志中fragment对应的是消息偏移量而非分区号，将上述代码中的record.partition()替换为record.offset()即可
• 优点：逻辑完全可控，输出格式可直接匹配需求，额外性能损耗可忽略

方案二：服务端审计日志采集 #

如果不方便修改消费者代码，可直接开启Kafka服务端审计日志，后续做格式清洗即可：

• 修改Kafka集群server.properties配置文件，新增如下配置：

# 开启审计日志
audit.log.enable=true
# 仅收集消费请求的审计日志，也可按需增加PRODUCE等其他请求类型
audit.log.event.types=FETCH,CONSUME

• 开启后Kafka会自动记录所有消费请求的客户端IP、请求topic、分区等核心字段，后续可通过日志清洗工具（如Logstash、Flink）将原生审计日志转换为你需要的目标格式
• 优点：无需修改任何消费者业务代码，全量采集无遗漏
• 缺点：原生审计日志字段冗余体积大，需要额外做格式转换，高并发集群下会增加服务端IO和存储压力

方案三：拦截器统一实现 #

如果不想分散修改每个消费者的业务逻辑，可通过拦截器统一植入日志逻辑：

• 消费者端拦截器：自定义实现ConsumerInterceptor接口，重写onConsume方法统一加日志输出，所有消费者仅需引入拦截器依赖、在配置中指定拦截器类即可，无需改动原有消费逻辑
• 服务端拦截器：自定义Kafka服务端拦截器，捕获所有FETCH请求后记录客户端IP、请求topic、分区等信息输出日志

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