Kafka

安全认证配置 SASL

常规配置(kafka和zk)

kafka server配置

#进入kafka的配置目录config
cd /mnt/data/kafka/config
#新建配置文件
touch kafka_server.jaas.conf
vim kafka_server.jaas.conf
#注意password后边有;
KafkaServer {
 ?  org.apache.kafka.common.security.plain.PlainLoginModule required
 ?  username="admin"
 ?  password="admin-secret"
 ?  user_admin="admin-secret";
};
KafkaClient {
 ?  org.apache.kafka.common.security.plain.PlainLoginModule required
 ?  username="admin"
 ?  password="admin-secret";
};
Client {
 ?  org.apache.kafka.common.security.plain.PlainLoginModule required
 ?  username="admin"
 ?  password="admin-secret";
};

修改server.properties

vim server.properties
?
# 允许外部端口连接
listeners=SASL_PLAINTEXT://0.0.0.0:9092
# 外部代理地址 外网地址
advertised.listeners=SASL_PLAINTEXT://101.xx.xx.xx:29092
?
security.inter.broker.protocol=SASL_PLAINTEXT
?
sasl.enabled.mechanisms=PLAIN
?
sasl.mechanism.inter.broker.protocol=PLAIN
?
# Add acl
allow.everyone.if.no.acl.found=true
auto.create.topics.enable=true
delete.topic.enable=true
advertised.host.name=dn1
super.users=User:admin
?
# Add class
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

修改kafka/bin

进入kafka的bin目录bin:
vim kafka-server-start.sh
?
export KAFKA_OPTS="-Djava.security.auth.login.config=/mnt/data/kafka/config/kafka_server_jaas.conf"
?

配置zookeeper

kafka_zoo.jaas.conf

touch kafka_zoo.jaas.conf
vim kafka_zoo.jaas.conf
ZKServer{
    org.apache.kafka.common.security.plain.PlainLoginModule required
        username="admin"
 ?      password="admin-secret"
 ?      user_kafka="admin-secret";
};
?

修改zookeeper-server-start.sh

export KAFKA_OPTS="-Djava.security.auth.login.config=/mnt/data/kafka/config/kafka_zoo_jaas.conf -Dzookeeper.allowSaslFailedClients=false"

如果需要使用kafka自带客户端需要配置

kafka client配置 kafka/config目录

touch kafka_client.jaas.conf
vim kafka_client.jaas.conf
Client {
 ?  org.apache.kafka.common.security.plain.PlainLoginModule required
 ?  username="admin"
 ?  password="admin-secret"
 ?  user_kafka="admin-secret";
};

修改producer.properties

vim producer.properties
?
security.protocol=SASL_PLAINTEXT
?
sasl.mechanism=PLAIN

修改consumer.properties

vim consumer.properties
?
security.protocol=SASL_PLAINTEXT
?
sasl.mechanism=PLAIN

修改启动脚本 kafka/bin目录

vim kafka-console-producer.sh
?
export KAFKA_OPTS="-Djava.security.auth.login.config=/mnt/data/kafka/config/kafka_client_jaas.conf"
?
vim kafka-console-consumer.sh
export KAFKA_OPTS="-Djava.security.auth.login.config=/mnt/data/kafka/config/kafka_client_jaas.conf"

kafka-java配置

1.properties配置

properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
properties.put(SaslConfigs.SASL_MECHANISM,"PLAIN");
properties.put(SaslConfigs.SASL_JAAS_CONFIG,"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin_secret\";");

kafka脚本

windows

#kafka\bin\windows目录
.\zookeeper-server-start.bat ..\..\config\zookeeper.properties
.\kafka-server-start.bat ..\..\config\server.properties
?

linux

#kafka/bin目录
#启动zookeeper
./zookeeper-server-start.sh -daemon /mnt/data/kafka/config/zookeeper.properties
#启动kafka
./kafka-server-start.sh -daemon /mnt/data/kafka/config/server.properties
#启动eagle
./ke.sh start
#查询topic信息
./kafka-topics.sh --describe --zookeeper localhost:2181 --topic test_topic1
#查询topic内容
./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test_topic1 --from-beginning