修改canal.properties配置文件，增加如下配置, 如果是使用了canal管理平台，在集群管理--》主配置里进行添加如下配置。 kafka认证机制为：PLAIN

kafka.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required  username='admin'  password='密碼';
kafka.sasl.mechanism=PLAIN
kafka.security.protocol=SASL_PLAINTEXT

kafka认证机制为：SCRAM-SHA-256

kafka.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username='admin'  password='密碼';
kafka.sasl.mechanism=SCRAM-SHA-256
kafka.security.protocol=SASL_PLAINTEXT

重启canal-server

cd bin/ && sh restart.sh
如果是canal的管理工具修改配置后自动重启。

参考：

https://www.jianshu.com/p/14cd7d64bb4e