查看哪些IP在向kafka的broker生产消息

问题来源

数据采集端正在调试，相关人员告知已经将所有的推送端服务都停止了，为何kafka上相关topic还能不断接收到消息，要求排查是哪些ip还在推送数据。

问了一个DeepSeek，得到的答案是Kafka本身并不直接记录或提供推送数据的客户端IP信息。Kafka的设计理念是追求高吞吐量和低延迟，因此没有内置记录客户端IP的功能。

提供了三种方案：

1、使用网络层工具：使用netstat或lsof命令来查看与Kafka broker建立的连接，方法简单直接，但缺点是这些IP地址无法与Kafka的客户端ID直接关联起来，而且信息是实时的，无法追溯历史记录。

2、临时修改Kafka日志级别：修改log4j.properties文件，将kafka.network.Processor的日志级别设置为DEBUG并重启，这样Kafka会在日志中记录连接细节，包括客户端的IP地址，不过搜索结果提到这种方法会增加磁盘I/O和存储需求，不建议长期使用。

3、使用抓包工具：可以建立客户端ID与IP的关联关系，这种方法需要交叉分析，但能提供更丰富的信息。

可选方案

结合实际情况，在不影响线上业务的情况下，快速查出正在向broker发送消息的ip即可，故选择第一个方案，在kafka broker节点上使用 

netstat -anp | grep :9092 | grep ESTABLISHED

或者

lsof -i :9092 | grep java

这里最终使用 netstat，但是打印出来的消息较多，里面包含集群中的其他broker、消费端的ip，结合grep -v排除集群的其他ip和消费端ip 和 awk确定打印指定列

最终方案

netstat -anp | grep :9092 | grep ESTABLISHED  | grep -v '消费端IP' | grep -v '集群中其他节点IP'  | awk '{print $5}'

通过以上命令就可以快速找到推送端ip地址了