如何以纯文本形式获取Kafka Consumer的消费结果

问题原因 #

默认启动的kafka-console-consumer会持续监听对应topic的新消息，进程不会主动退出，属于常驻交互进程，因此无法被仅支持捕获已结束命令返回值的Robot Framework SSHLibrary采集消费结果。

核心修改方式 #

给kafka-console-consumer追加进程自动退出相关参数，让消费完成后进程自动终止，所有消费到的消息会作为标准输出直接返回，适配自动化采集要求：

• --timeout-ms <毫秒值>：配置消费者连续指定时长内没有拉取到新消息时，自动退出进程。该参数的计时从最后一次成功拉取消息开始计算，不会截断符合消费位点要求的存量消息。
• --max-messages <消息条数>：配置消费者累计消费到指定条数的消息后立刻主动退出，适合已知待消费消息总量的场景，避免无效等待。

调整后可直接用于自动化的命令示例 #

非Kafka逻辑保持原有写法不变，仅追加Kafka消费端参数，同时去掉kubectl的交互式tty参数避免控制字符污染文本输出：

kubectl exec -i -n me2 `kubectl get po -n me2 | grep central | awk '{print $1}' | head -1` -- /usr/bin/kafka-console-consumer --bootstrap-server me-ckaf-kafka-headless:9092 --topic CdrToUsageRepRating --from-beginning --timeout-ms 5000

上述示例中--timeout-ms 5000代表连续5秒拉取不到新消息就自动退出，可根据集群网络延迟、topic消息生产速度调整该值，确保存量消息能被完整拉取。

适配注意点 #

• kafka-console-consumer默认逐行以纯文本格式打印消息内容，进程退出后所有输出可直接被SSHLibrary的命令执行方法捕获，不需要额外做格式转换。
• 自动化场景下不要保留kubectl的-t（分配tty终端）参数，否则输出会携带终端控制字符，干扰纯文本结果解析。
• 如果topic历史消息量较大，可搭配--max-messages参数设置消费条数上限，避免消费时间过长阻塞自动化用例执行。

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