canal 连接开启sasl验证的Kafka集群

底层逻辑：

1）Kafka开启了sasl验证，意味着客户端需要通过sasl验证才能连接Kafka。

2）客户端需要准备两样东西

      第一：配置 -Djava.security.auth.login.config

      第二：准备sasl验证文件

3）针对canal来讲，需要开启sasl验证模式。

具体操作如下：

1）canal.properties文件修改，用来指定数据发送kafka  

   canal.serverMode = kafka

2）canal.properties文件新增两行配置，用来开启sasl验证模式。

    canal.mq.properties.security.protocol = SASL_PLAINTEXT
     canal.mq.properties.sasl.mechanism = PLAIN

3）instance.properties 文件修改 canal.mq.topic=maoxiangyi，用来指定数据发往哪个kafka

KafkaClient {
         org.apache.kafka.common.security.plain.PlainLoginModule required
         username="sss"
         password="ssss";
};

4）准备一个文件kafka_client_jaas.conf，放在config目录下。

5）修改启动脚本 start.sh ，配置 -Djava.security.auth.login.config=/opt/canal/conf/kafka_client_jaas.conf。 具体位置参考以下代码

if [ -n "$str" ]; then
     JAVA_OPTS="-server -Xms2048m -Xmx3072m -Xmn1024m -XX:SurvivorRatio=2 -XX:PermSize=96m -XX:MaxPermSize=256m -Xss256k -XX:-UseAdaptiveSizePolicy -XX:MaxTenuringThreshold=15 -XX:+DisableExplicitGC -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseFastAccessorMethods -XX:+UseCMSInitiatingOccupancyOnly -XX:+HeapDumpOnOutOfMemoryError -Djava.security.auth.login.config=/opt/canal/conf/kafka_client_jaas.conf "
else
     JAVA_OPTS="-server -Xms1024m -Xmx1024m -XX:NewSize=256m -XX:MaxNewSize=256m -XX:MaxPermSize=128m "
fi

打完收工。

kafka 开启sasl验证，参考  https://blog.csdn.net/u010416101/article/details/79562214?utm_source=blogxgwz0