在monitor_nginx.conf 加入下面配置

logser.sinks.kafka_monitor_nginx.kafka.producer.security.protocol = SASL_PLAINTEXT
logser.sinks.kafka_monitor_nginx.kafka.producer.sasl.mechanism = PLAIN

cp kafka的lib内jar到flume的lib下  cp以下包

kafka-clients-2.5.0.jar
lz4-java-1.7.1.jar
osgi-resource-locator-1.0.1.jar
slf4j-api-1.7.30.jar
snappy-java-1.1.7.3.jar

创建验证文件 flume_client_jass.conf

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

 修改启动文件flume-ng  加入以下内容到文件最后

export JAVA_OPTS="-Djava.security.auth.login.config=/usr/local/apache-flume-1.9.0-bin/conf/flume_client_jass.conf"

 flume_client_jass.conf 配置内容是和kafka 自定义的验证内容 不熟悉可以看看上篇文章