如何借助Kafka Connect与XML配置文件监控Kafka Topic内容？

一、XML配置转Kafka Connect Properties格式 #

原生Apache Kafka Connect不使用XML配置文件，你拿到的XML是第三方Kafka管理工具（比如集群管理界面）的连接配置，只需提取核心安全参数，添加到你的connect-standalone.properties中即可：

需要添加的安全配置项 #

在现有配置下方追加以下内容，替换XML里的USER和PASSWD为实际账号密码：

# 对应XML中的broker_security_type="SASL_SSL"
security.protocol=SASL_SSL
# 对应XML中的sasl_mechanism="SCRAM-SHA-512"
sasl.mechanism=SCRAM-SHA-512
# 对应XML中的jaas_config，注意转义引号改为双引号
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="USER" password="PASSWD";

无需处理的XML字段 #

XML里的groupId、group、name、chroot等是第三方工具的自定义参数，原生Kafka Connect不需要，直接忽略；keystore、truststore相关值为空，也无需配置。

二、启动本地独立消费者查看Topic内容 #

不用依赖Kafka Connect，直接用Kafka自带的脚本就能快速监控Topic内容，步骤如下：

1. 创建消费者安全配置文件：在Kafka目录下新建consumer.properties，写入安全认证信息：

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="USER" password="PASSWD";

2. 启动控制台消费者：执行以下命令，替换YOUR_TOPIC_NAME为你要监控的Topic名称：

bin/kafka-console-consumer.sh \
--bootstrap-servers xxxxxx.c3.kafka.eu-west-3.amazonaws.com:9096,xxxxxx.c3.kafka.eu-west-3.amazonaws.com:9096,xxxxxx.c3.kafka.eu-west-3.amazonaws.com:9096 \
--topic YOUR_TOPIC_NAME \
--from-beginning \
--consumer.config consumer.properties

• --from-beginning参数会读取Topic中所有历史消息，若只需监控新消息，去掉该参数即可。

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