本主要介绍在 Kafka 中如何配置 Kerberos 认证,文中所使用到的软件版本:Java 1.8.0_191、Kafka 2.13-2.4.1、Kerberos 1.15.1,
1、Kerberos 安装
要使用 Kerberos 服务,需先安装 Kerberos,安装方法可参考:Kerberos 入门实战(2)--Kerberos 安装及使用,
2、Zookeeper 开启 Kerberos 认证
Zookeeper 开启 Kerberos 认证的方法参见:Zookeeper 入门实战(4)--开启 Kerberos 认证,
3、Kafka 开启 Kerberos 认证
假设 Kafka 集群信息如下:
ip | 主机名 | 描述 |
10.49.196.10 | pxc1 | zookeeper、kafka |
10.49.196.11 | pxc2 | zookeeper、kafka |
10.49.196.12 | pxc3 | zookeeper、kafka |
3.1、创建 keytab
在安装 Kerberos 的机器上进入 kadmin(Kerberos 服务端上使用 kadmin.local,安装了 Kerberos Client 的机器上可以使用 kadmin),然后执行如下命令分别创建服务端和客户端的 keytab:
kadmin.local: add_principal -randkey kafka-server/pxc1@ABC.COM kadmin.local: add_principal -randkey kafka-server/pxc2@ABC.COM kadmin.local: add_principal -randkey kafka-server/pxc3@ABC.COM kadmin.local: add_principal -randkey kafka-client@ABC.COM kadmin.local: xst -k /root/zk-server.keytab kafka-server/pxc1@ABC.COM kadmin.local: xst -k /root/zk-server.keytab kafka-server/pxc2@ABC.COM kadmin.local: xst -k /root/zk-server.keytab kafka-server/pxc3@ABC.COM kadmin.local: xst -k /root/zk-client.keytab kafka-client@ABC.COM
3.2、配置
3.2.1、配置 hosts 档案
10.49.196.10 pxc1 10.49.196.11 pxc2 10.49.196.12 pxc3
3.2.2、Kerberos 相关配置
拷贝 krb5.conf 及 keytab 档案到所有安装 Kafka 的机器,这里把档案都放到 Kafka 的 config/kerveros 目录下(kerberos 目录需新建),
2.2.3、Kafka 服务端配置
A、修改 config/server.properties,增加如下配置:
security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=GSSAPI sasl.enabled.mechanisms=GSSAPI sasl.kerberos.service.name=kafka-server
B、新建 kafka-server-jaas.conf 档案,该档案也放到 Kafka 的 config/kerveros 目录下,
KafkaServer { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/home/hadoop/app/kafka_2.13-2.4.1/config/kerberos/kafka-server.keytab" storeKey=true useTicketCache=false principal="kafka-server/pxc1@ABC.COM"; #不同的主机,需修改为本机的主机名 }; //Zookeeper client authentication Client { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/home/hadoop/app/kafka_2.13-2.4.1/config/kerberos/kafka-server.keytab" storeKey=true useTicketCache=false principal="kafka-server/pxc1@ABC.COM"; #不同的主机,需修改为本机的主机名 };
C、修改 bin/kafka-server-start.sh 脚本,倒数第二行增加如下配置:
export KAFKA_OPTS="-Dzookeeper.sasl.client=true -Dzookeeper.sasl.client.username=zk-server -Djava.security.krb5.conf=/home/hadoop/app/kafka_2.13-2.4.1/config/kerberos/krb5.conf -Djava.security.auth.login.config=/home/hadoop/app/kafka_2.13-2.4.1/config/kerberos/kafka-server-jaas.conf"
如果 Zookeeper 没有开启 Kerberos 认证,则这里 zookeeper.sasl.client 可设为 false;仅仅启用 Kafka 的 Kerberos 认证,
3.2.4、Kafka 客户端配置
该配置主要为了使用 bin/kafka-topics.sh、bin/kafka-console-consumer.sh、kafka-console-producer.sh 等命令,
A、新建 client.properties 档案,该档案也放到 Kafka 的 config/kerveros 目录下,
security.protocol=SASL_PLAINTEXt sasl.mechanism=GSSAPI sasl.kerberos.service.name=kafka-server
B、新建 kafka-client-jaas.conf 档案,该档案也放到 Kafka 的 config/kerveros 目录下,
KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="/home/hadoop/app/kafka_2.13-2.4.1/config/kerberos/kafka-client.keytab" storeKey=true useTicketCache=false principal="kafka-client@ABC.COM"; };
C、修改 bin/kafka-topics.sh、bin/kafka-console-consumer.sh、kafka-console-producer.sh 脚本,倒数第二行增加如下配置:
export KAFKA_OPTS="-Djava.security.krb5.conf=/home/hadoop/app/kafka_2.13-2.4.1/config/kerberos/krb5.conf -Djava.security.auth.login.config=/home/hadoop/app/kafka_2.13-2.4.1/config/kerberos/kafka-client-jaas.conf"
3.3、启动并测验
配置完成后就可以启动 Kafka 集群了:
cd /home/hadoop/app/kafka_2.13-2.4.1/bin
./kafka-server-start.sh -daemon ../config/server.properties
启动完成后可以使用 bin/kafka-topics.sh、bin/kafka-console-consumer.sh、kafka-console-producer.sh 来测验:
查看 topic 信息:
cd /home/hadoop/app/kafka_2.13-2.4.1/bin ./kafka-topics.sh --list --bootstrap-server 10.49.196.10:9092 --command-config ../config/kerberos/client.properties
发送讯息:
cd /home/hadoop/app/kafka_2.13-2.4.1/bin ./kafka-console-producer.sh --broker-list 10.49.196.10:9092 --topic test --producer.config ../config/kerberos/client.properties
接受讯息:
cd /home/hadoop/app/kafka_2.13-2.4.1/bin ./kafka-console-consumer.sh --bootstrap-server 10.49.196.10:9092 --topic test --from-beginning --consumer.config ../config/kerberos/client.properties
3.4、java 程序连接 Zookeeper
java 可以使用 JAAS 来进行 Kerberos 认证,需要 JAAS 组态档、keytab 档案及 Kerberos 组态档,
A、组态档
JAAS 组态档(kafka-client-jaas.conf):
KafkaClient { com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true keyTab="D:\\workspaceidea\\demo\\demo\\src\\main\\resources\\kerberos\\kafka-client.keytab" storeKey=true useTicketCache=false principal="kafka-client@ABC.COM"; };
keytab 档案:
从 Kerberos 服务器上拷贝到目标机器,拷贝路径即为 JAAS 配置中间配置的路径:D:\\workspaceidea\\demo\\demo\\src\\main\\resources\\kerberos\\kafka-client.keytab,
Kerberos 组态档(krb5.conf):
从 Kerberos 服务器上拷贝 /etc/krb5.conf 到目标机器即可,
B、配置 hosts 档案
在 hosts 档案中添加:
10.49.196.10 pxc1 10.49.196.11 pxc2 10.49.196.12 pxc3
C、引入依赖
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.4.1</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-streams</artifactId> <version>2.4.1</version> </dependency>
D、样例程序
package com.inspur.demo.general.kafka; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.ListTopicsOptions; import org.apache.kafka.clients.admin.ListTopicsResult; import org.apache.kafka.clients.admin.TopicListing; import org.junit.After; import org.junit.Before; import org.junit.Test; import java.util.Collection; import java.util.Properties; public class KafkaKerberos { private AdminClient adminClient; @Before public void before() { System.setProperty("java.security.auth.login.config", "D:\\workspaceidea\\demo\\demo\\src\\main\\resources\\kerberos\\kafka-client-jaas.conf"); System.setProperty("java.security.krb5.conf", "D:\\workspaceidea\\demo\\demo\\src\\main\\resources\\kerberos\\krb5.conf"); Properties props = new Properties(); props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "10.49.196.10:9092,10.49.196.11:9092,10.49.196.12:9092"); props.put("sasl.mechanism", "GSSAPI"); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.kerberos.service.name", "kafka-server"); adminClient = AdminClient.create(props); } @After public void after() { adminClient.close(); } @Test public void listTopics() throws Exception { ListTopicsOptions listTopicsOptions = new ListTopicsOptions(); //是否罗列内部主题 listTopicsOptions.listInternal(true); ListTopicsResult result = adminClient.listTopics(listTopicsOptions); Collection<TopicListing> list = result.listings().get(); System.out.println(list); } }
0 评论