后台修改集群配置

1. 修改主master节点的/opt/Bigdata/components/FusionInsight_HD_8.1.0.1/Kafka/configurations.xml

2. 搜索sasl.port，删除model="Sec"

3. 搜索sasl-ssl.port，删除model="Sec"

4. 搜索listeners，修改1508行，将model="Sec"删除，将1509行注释

5. 搜索security.inter.broker.protocol，修改1363，将model="Sec"删除，将1362行注释

6. 搜索sasl.mechanism.inter.broker.protocol，将type="hidden"修改为type="advanced"

7. 搜索authorizer.class.name，将value一行改为<value vType="string"> kafka.security.authorizer.AclAuthorizer </value>

8. Root切换成omm用户后，执行命令重启controller

su - omm

sh /opt/Bigdata/om-server/om/sbin/restart-controller.sh

9. 在集群所有部署kafka broker节点的机器创建文件

/opt/Bigdata/kafka_server_scram_jaas.conf

文件内容为：

KafkaServer {

org.apache.kafka.common.security.scram.ScramLoginModule required

username="kafka"

password="kafka-secret";

};

kafka-secret为kafka用户密码，可自定义，修改后创建用户命令中的密码需同步修改

修改FusionInsight Manager页面配置

在FusionInsight Manager页面，修改Kafka配置

1. KAFKA_JVM_PERFORMANCE_OPTS后追加“-Djava.security.auth.login.config=/opt/Bigdata/kafka_server_scram_jaas.conf”，注意与前面的内容用空格隔开

2. 修改配置项sasl.mechanism.inter.broker.protocol值为SCRAM-SHA-256

3. 修改配置项sasl.enabled.mechanisms值为SCRAM-SHA-256

4. 添加自定义配置allow.everyone.if.no.acl.found = true

5. 添加自定义配置is.security.mode = yes

kafka服务同步配置后重启服务

创建用户

重新下载客户端并安装：https://support.huaweicloud.com/usermanual-mrs/mrs_01_0090.html （目前安装在/opt/client下）

登录任一客户端bin目录，执行命令创建用户

1. 执行命令创建kafka super user

./kafka-configs.sh --zookeeper node-master2ZgUf:2181/kafka --alter --add-config 'SCRAM-SHA-256=[password=kafka-secret],SCRAM-SHA-512=[password=kafka-secret]' --entity-type users --entity-name kafka

2. 执行命令创建业务用户test

./kafka-configs.sh --zookeeper node-master2ZgUf:2181/kafka --alter --add-config 'SCRAM-SHA-256=[password=test-secret],SCRAM-SHA-512=[password=test-secret]' --entity-type users --entity-name test

3. 在每个broker节点添加系统用户：useradd test

创建topic并授权

./kafka-topics.sh --create --topic topicname --replication-factor 1 --partitions 1 --zookeeper  node-master2ZgUf:2181/kafka

./kafka-acls.sh --authorizer-properties zookeeper.connect= node-master2ZgUf:2181/kafka --add --allow-principal User:test --operation Read --operation Write --topic topicname

验证

1. 修改生产者配置文件/opt/client/Kafka/kafka/config/producer.properties

将security.protocol 修改为SASL_PLAINTEXT，并添加如下两行

sasl.mechanism=SCRAM-SHA-256

sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \

username="test" \

password="test-secret";

2. 修改消费者配置文件/opt/client/Kafka/kafka/config/consumer.properties

将security.protocol修改为SASL_PLAINTEXT，并添加如下两行

sasl.mechanism=SCRAM-SHA-256

sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \

username="test" \

password="test-secret";

3. 执行命令验证生产者：使用21007端口

kafka-console-producer.sh --broker-list 192.168.5.140:21007,192.168.5.183:21007,192.168.5.210:21007 -topic topicname --producer.config /opt/client/Kafka/kafka/config/producer.properties

4. 执行命令验证消费者：

kafka-console-consumer.sh --bootstrap-server 172.16.0.45:21007,172.16.0.201:21007,172.16.0.15:21007 --topic topicname --consumer.config /opt/client/Kafka/kafka/config/consumer.properties