1.在kerberos管理节点的root用户下分别执行以下命令
[root@nyyjh1 ~]#kadmin.localAuthenticating as principal root/admin@ with password.kadmin.local:addprinc-randkeykafka/nyyjh2@kadmin.local:addprinc-randkeykafka/nyyjh3@kadmin.local:addprinc-randkeykafka/nyyjh4@kadmin.local:addprinc-randkeyzookeeper/nyyjh2@kadmin.local:addprinc-randkeyzookeeper/nyyjh3@kadmin.local:addprinc-randkeyzookeeper/nyyjh4@kadmin.local:xst-kkafka.keytabkafka/nyyjh2@kadmin.local:xst-kkafka.keytabkafka/nyyjh3@kadmin.local:xst-kkafka.keytabkafka/nyyjh4@exit[root@nyyjh1 ~]#scp kafka.keytab root@nyyjh2:/etc/kafka1/conf/[root@nyyjh1 ~]#scp kafka.keytab root@nyyjh3:/etc/kafka1/conf/[root@nyyjh1 ~]#scp kafka.keytab root@nyyjh4:/etc/kafka1/conf/
2.配置Kafka各个节点的server.properties文件
listeners=SASL_PLAINTEXT://nyyjh2:9092advertised.listeners=SASL_PLAINTEXT://nyyjh2:9092sasl.enabled.mechanisms=GSSAPIsecurity.inter.broker.protocol=SASL_PLAINTEXTsasl.kerberos.service.name=kafkaauthorizer.class.name=kafka.security.auth.SimpleAclAuthorizersuper.users=User:kafka
3.配置Kafka各个节点的consumer.properties文件
key.deserializer=org.mon.serialization.StringDeserializervalue.deserializer=org.mon.serialization.StringDeserializerkey.serializer=org.mon.serialization.StringSerializervalue.serializer=org.mon.serialization.StringSerializersecurity.protocol=SASL_PLAINTEXTsasl.mechanism=GSSAPIsasl.kerberos.service.name=kafka
4.配置Kafka各个节点的producer.properties文件
key.deserializer=org.mon.serialization.StringDeserializervalue.deserializer=org.mon.serialization.StringDeserializerkey.serializer=org.mon.serialization.StringSerializervalue.serializer=org.mon.serialization.StringSerializersecurity.protocol=SASL_PLAINTEXTsasl.mechanism=GSSAPIsasl.kerberos.service.name=kafka
5.配置Kafka各个节点的bin/kafka-run-class.sh
增加KAFKA_KERBEROS_OPTS配置项，其中配置kerberos配置文件和kafka认证文件。并将该参数添加到启动时候的$JAVA后边。
# JVM performance optionsif [ -z "$KAFKA_JVM_PERFORMANCE_OPTS" ]; thenKAFKA_JVM_PERFORMANCE_OPTS="-server -XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+ExplicitGCInvokesConcurrent -Djava.awt.headless=true -Djava.security.krb5.conf=/etc/kafka1/conf/krb5.conf -Djava.security.auth.login.config=/etc/kafka1/conf/jaas.conf"fi
6.配置Kafka各个节点jaas.conf文件
KafkaServer {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truestoreKey=truekeyTab="/etc/kafka1/conf/kafka.keytab"principal="kafka/节点名称@"; };// ZooKeeper client authenticationKafkaClient {com.sun.security.auth.module.Krb5LoginModule requireduseKeyTab=truestoreKey=truekeyTab="/etc/kafka1/conf/kafka.keytab"principal="kafka/节点名称@";};
7.配置Kafka各个节点krb5.conf文件
[logging]default = FILE:/var/log/krb5libs.logkdc = FILE:/var/log/krb5kdc.logadmin_server = FILE:/var/log/kadmind.log[libdefaults]dns_lookup_realm = falseticket_lifetime = 24hrenew_lifetime = 7dforwardable = truerdns = falsepkinit_anchors = FILE:/etc/pki/tls/certs/ca-bundle.crtdefault_realm = default_ccache_name = KEYRING:persistent:%{uid}[realms] = {kdc = nyyjh1admin_server = nyyjh1}[domain_realm]# . = # =
8.启动Kafka测试
/usr/local/kafka/bin/kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
9.创建topic测试
kafka-topics.sh --create --partitions 1 --replication-factor 3 --zookeeper nyyjh1:2181,nyyjh2:2181,nyyjh3:2181 --topic testKafka1
10.终端测试
#启动生产端kafka-console-producer.sh --broker-list nyyjh2:9092,nyyjh3:9092,nyyjh4:9092 --topic testKafka --producer.config /usr/local/kafka/config/producer.properties#启动消费端kafka-console-consumer.sh --bootstrap-server nyyjh2:9092,nyyjh3:9092,192.168.122.104:9092 --consumer.config /usr/local/kafka/config/consumer.properties --topic testKafka --from-beginning
完成！
如果觉得《Kafka配置kerberos安全认证》对你有帮助，请点赞、收藏，并留下你的观点哦！
