如何连接Kafka并复制数据？持给定参数用Conduktor连接报认证错误怎么办

第一步：确认集群认证类型 #

首先需要向客户确认Kafka集群使用的认证协议，不同协议的配置逻辑完全不同，自行猜测参数名无法完成配置。常见的Kafka认证类型包括SASL/PLAIN、SASL/SCRAM、SSL双向认证三类。

第二步：按认证类型配置连接参数 #

场景1：SASL/PLAIN、SASL/SCRAM认证（账号密码模式最常见） #

• 打开Conduktor新建集群配置页，将客户提供的bootstrap.servers地址填入对应输入框
• 切换到「Security」标签页，Security protocol选择SASL_PLAINTEXT（集群未开启SSL传输加密）或者SASL_SSL（集群开启SSL传输加密）
• SASL mechanism选择对应认证类型：PLAIN对应SASL/PLAIN，SCRAM-SHA-256/SCRAM-SHA-512对应SASL/SCRAM
• 直接将客户提供的账号填入Username输入框、密码填入Password输入框即可，不需要修改参数名也不需要加额外前缀

场景2：SSL双向认证 #

• Security protocol选择SSL
• 不需要填写SASL类用户名密码，上传客户提供的CA根证书、客户端证书、客户端密钥文件，按提示输入密钥库密码即可

第三步：排错验证 #

• 配置完成后先点击「Test connection」验证连通性，如果仍返回认证错误，先用Kafka官方命令行工具验证账号有效性，排除账号密码本身错误的问题：

示例SASL_SSL+PLAIN认证的命令行测试配置：
kafka-console-consumer.sh --bootstrap-server 集群地址 --topic 测试Topic --consumer-property security.protocol=SASL_SSL --consumer-property sasl.mechanism=PLAIN --consumer-property sasl.jaas.config='org.apache.kafka.common.security.plain.PlainLoginModule required username="你的账号" password="你的密码";'

• 确认本地网络可以连通bootstrap.servers对应所有节点的服务端口，排除防火墙、白名单限制
• 确认所用账号拥有目标Topic的读写权限，部分集群权限配置错误也会返回类认证错误提示

数据复制操作 #

• 连接验证通过后，可直接使用Conduktor内置的Topic数据复制功能，选择对应源Topic、目标Topic即可完成数据复制需求。

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