Kafka添加Acl认证

一、Kafka安装

#### 1. 在安装Kafka的基础配置上添加或修改以下参数
$KAFKA_HOME/config/server.properties
listeners=SASL_PLAINTEXT://hadoop02:9092

如果设置此参数在集群状态下启动会失败

allow.everyone.if.no.acl.found=true
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer

超级用户

super.users=User:admin
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

2. 修改Kafka启动脚本添加以下内容

$KAFKA_HOME/bin/kafka-server-start.s
export KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS -Djava.security.auth.login.config=$KAFKA_HOME/jass/kafka-server-jaas.conf"

3. 创建kafka-server-jass.conf配置文件

$KAFKA_HOME/jass/kafka-server-jaas.conf

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin"
    user_admin="admin"
    user_writer="writer"
    user_reader="reader";
};

此文件做认证使用 格式为： user_用户=密码

二、配置Producter（生产者）

1. 修改Kafka生产者脚本 添加以下内容

$KAFKA_HOME/bin/kafka-console-producer.sh

KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS -Djava.security.auth.login.config=$KAFKA_HOME/jass/kafka-writer-jass.conf"

2. 创建生产者jass文件

$KAFKA_HOME/jass/kafka-writer-jass.conf
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="writer"
password="writer";
};

image.png

3. 创建生产者脚本启动需要的配置文件

$KAFKA_HOME/jass/producer.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

三、创建测试主题测试生产者

1. 创建wxgz主题

kafka-topics.sh --create --zookeeper hadoop02:2181 --topic wxgz --partitions 1 --replication-factor 1

创建成功（注意正常来说加了验证是不应该创建成功的，但是kafka-topic.sh脚本是直接和zookeeper交互的，
不经过kafka所以创建成功，也就是说kafka-topic.sh脚本不受acl的控制）

2. 给用户授权

kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:writer --operation Write --topic wxgz

3. 测试 向wxgz 主题中生产数据

kafka-console-producer.sh --broker-list hadoop02:9092 --topic wxgz --producer.config jass/producer.conf

写入成功

四、配置和测试消费者

1. 修改消费者脚本 添加一下内容

$KAFKA_HOME/bin/kafka-console-consumer.sh
KAFKA_HEAP_OPTS="$KAFKA_HEAP_OPTS -Djava.security.auth.login.config=$KAFKA_HOME/jass/kafka-reader-jass.conf"

2. 创建消费者jass配置文件

$KAFKA_HOME/jass/kafka-reader-jass.confsecurity.protocol=SASL_PLAINTEXT
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="reader"
password="reader";
};

3. 创建消费者脚本启动需要的配置文件

$KAFKA_HOME/jass/consumer.conf
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
group.id=test-group

4. 授予用户读权限

kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:reader --operation Read --topic wxgz

5. 测试消费数据

kafka-console-consumer.sh --bootstrap-server hadoop02:9092 --topic wxgz --from-beginning --consumer.config jass/consumer.conf

五、常用命令

#### 1.创建主题
kafka-topics.sh --create --zookeeper hadoop02:2181 --topic wxgz --partitions 1 --replication-factor 1

#### 2. 授予用户写权限
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:writer --operation Write --topic wxgz

#### 3. 启动生产者
kafka-console-producer.sh --broker-list hadoop02:9092 --topic wxgz --producer.config jass/producer.conf

#### 4. 授予用户读权限
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:reader --operation Read --topic wxgz

#### 5. 授予消费组读权限
kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=hadoop02:2181 --add --allow-principal User:reader --operation Read --group test-group

#### 6. 启动消费者
kafka-console-consumer.sh --bootstrap-server hadoop02:9092 --topic wxgz --from-beginning --consumer.config jass/consume.conf