C/C++教程
大数据Hadoop之——Kafka鉴权认证（Kafka kerberos认证+kafka账号密码认证+CDH Kerberos认证）
本文主要是介绍大数据Hadoop之——Kafka鉴权认证（Kafka kerberos认证+kafka账号密码认证+CDH Kerberos认证），对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！
目录- 一、Kerberos概述与安装
- 二、Kafka Kerberos认证
- 1）创建用户和生成用户keytab
- 2）Kerberos 相关配置
- 3）Kafka 服务端配置
- 1、修改server.properties
- 2、修改kafka-server-start.sh
- 4）Kafka 客户端配置
- 1、配置client.properties
- 2、配置kafka-client-jaas.conf
- 3、配置kafka-topics.sh等操作文件
- 5）将配置文件copy到其它机器上
- 6）测试验证
- 1、查看topic 列表
- 2、创建topic
- 3、查看topic详情
- 4、删除topic
- 4、生产/消费
- 5、查看消费组积压
- 四、Kafka 用户密码认证（SASL/PLAIN ）
- 1）Kafka 服务端配置
- 1、配置jaas
- 2、配置服务启动参数（kafka-server-start.sh）
- 3、配置服务启动文件（server.properties）
- 2）Kafka 客户端配置
- 1、创建客户端认证配置文件
- 2、在配置文件中添加属性
- 3、配置kafka-topics.sh等操作文件
- 3）测试验证
- 1、查看topic列表
- 2、创建topic
- 3、查看topic详情
- 4、删除topic
- 4、生产/消费
- 5、查看消费组积压
- 1）Kafka 服务端配置
- 五、CDH 启用Kerberos认证
- 1）安装Kerberos
- 2）为CM创建管理员主体/实例
- 3）在CM上开启Kerberos认证
- 3）简单使用
- 1、在Kerberos数据库中创建用户主体/实例
- 2、进行用户认证访问HDFS
一、Kerberos概述与安装
Kerberos概述与安装可以参考我之前的文章：Kerberos认证原理与环境部署
二、Kafka Kerberos认证
Kafka安装可以参考我以前的文章：
大数据Hadoop之——Kafka 图形化工具 EFAK（EFAK环境部署）
分布式开源协调服务——Zookeeper
Kafka Kerberos认证官方文档：https://kafka.apache.org/31/documentation.html#security_sasl
Kafka 是服务，所以这里使用的是服务principal，不清楚的小伙伴可以看我上面的文章。格式如下：
Service/Hostname@REALM
【温馨提示】
SASL：简单验证和安全层 (Simple Authentication and Security Layer,)
1）创建用户和生成用户keytab
在安装 Kerberos 的机器上进入 kadmin（Kerberos 服务端上使用 kadmin.local，安装了 Kerberos Client 的机器上可以使用 kadmin），然后执行如下命令分别创建服务端和客户端的 keytab：
# 创建服务端的keytab，在不同的机器上使用不同主机名对应的keytab文件，格式：principal/hostname@realm # 【温馨提示】kafka-server要与下面的sasl.kerberos.service.name一致。 $ kadmin.local -q "add_principal -randkey kafka-server/hadoop-node1@HADOOP.COM" # 创建客户端的keytab（hadoop-node2） $ kadmin.local -q "add_principal -randkey kafka-server/hadoop-node2@HADOOP.COM" # 创建客户端的keytab（hadoop-node3） $ kadmin.local -q "add_principal -randkey kafka-server/hadoop-node3@HADOOP.COM" # 创建客户端用户 $ kadmin.local -q "add_principal -randkey kafka-client@HADOOP.COM" # 导出keytab文件，注意这里只能是kafka-server.keytab名字，换成其它名字会报错。这里为了区分，先起另外的名字，在其它节点的时候修改回kafka-server.keytab $ kadmin.local -q "xst -k /root/kafka-server.keytab kafka-server/hadoop-node1@HADOOP.COM" $ kadmin.local -q "xst -k /root/kafka-server-node2.keytab kafka-server/hadoop-node2@HADOOP.COM" $ kadmin.local -q "xst -k /root/kafka-server-node3.keytab kafka-server/hadoop-node3@HADOOP.COM" $ kadmin.local -q "xst -k /root/kafka-client.keytab kafka-client@HADOOP.COM" # 验证 $ kinit -kt /root/kafka-server.keytab kafka-server/hadoop-node1 $ kinit -kt /root/kafka-server-node2.keytab kafka-server/hadoop-node2 $ kinit -kt /root/kafka-server-node3.keytab kafka-server/hadoop-node3 $ kinit -kt /root/kafka-client.keytab kafka-client
2）Kerberos 相关配置
拷贝 krb5.conf 及kafka.keytab 文件都放到 Kafka 的 $KAFKA_HOME/config/kerberos 目录下(kerberos 目录需新建)。
$ mkdir $KAFKA_HOME/config/kerberos $ cp /root/kafka*.keytab $KAFKA_HOME/config/kerberos/ # 将krd5.conf也复制到$KAFKA_HOME/config/kerberos目录下 $ cp /etc/krb5.conf $KAFKA_HOME/config/kerberos/
3）Kafka 服务端配置
1、修改server.properties
复制 $KAFKA_HOME/config/server.properties（复制动作能帮助我们在 认证/非认证 模式自由切换）重命名为 $KAFKA_HOME/config/server-sasl.properties
$ scp $KAFKA_HOME/config/server.properties $KAFKA_HOME/config/server-sasl.properties
增加或修复如下配置：
# 【温馨提示】其中sasl.kerberos.service.name要和principal的名称相同 listeners=SASL_PLAINTEXT://0.0.0.0:19092 advertised.listeners=SASL_PLAINTEXT://hadoop-node1:19092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=GSSAPI sasl.enabled.mechanisms=GSSAPI sasl.kerberos.service.name=kafka-server
新建 kafka-server-jaas.conf 文件，该文件也放到 Kafka 的 $KAFKA_HOME/config/kerberos/ 目录下：
$ cat > $KAFKA_HOME/config/kerberos/kafka-server-jaas.conf <<EOF
KafkaServer {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
//不同的主机，需修改成不同的keytab文件
keyTab="/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/kerberos/kafka-server.keytab"
storeKey=true
useTicketCache=false
//不同的主机，需修改为本机的主机名
principal="kafka-server/hadoop-node1@HADOOP.COM";
};
//Zookeeper client authentication，配置跟KafkaServer 一样，如果zookeeper开启了kerberos认证，就得加这段配置，我这没开，所以先注释了
//Client {
// com.sun.security.auth.module.Krb5LoginModule required
// useKeyTab=true
// 不同的主机，需修改成不同的keytab文件
// keyTab="/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/kerberos/kafka-server.keytab"
// storeKey=true
// useTicketCache=false
// 不同的主机，需修改为本机的主机名
// principal="kafka-server/hadoop-node1@HADOOP.COM";
//};
EOF
2、修改kafka-server-start.sh
复制 bin/kafka-server-start.sh 脚本重命名为 bin/kafka-server-start-sasl.sh
$ cp $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/bin/kafka-server-start-sasl.sh
在倒数第二行增加如下配置（bin/kafka-server-start-sasl.sh）：
export KAFKA_OPTS="-Dzookeeper.sasl.client=false -Dzookeeper.sasl.client.username=zk-server -Djava.security.krb5.conf=/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/kerberos/krb5.conf -Djava.security.auth.login.config=/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/kerberos/kafka-server-jaas.conf"
【温馨提示】如果 Zookeeper 开启 Kerberos 认证，则这里
zookeeper.sasl.client就得设为 true。
4）Kafka 客户端配置
1、配置client.properties
该配置主要为了使用 bin/kafka-topics.sh、bin/kafka-console-consumer.sh、kafka-console-producer.sh 等命令。
新建 client.properties 文件，该文件也放到 Kafka 的 $KAFKA_HOME/config/kerberos 目录下：
$ cat > $KAFKA_HOME/config/kerberos/client.properties <<EOF security.protocol=SASL_PLAINTEXT sasl.mechanism=GSSAPI sasl.kerberos.service.name=kafka-server EOF
2、配置kafka-client-jaas.conf
新建 kafka-client-jaas.conf 文件，该文件也放到 Kafka 的 config/kerberos 目录下：
$ cat > $KAFKA_HOME/config/kerberos/kafka-client-jaas.conf <<EOF
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/kerberos/kafka-client.keytab"
storeKey=true
useTicketCache=false
principal="kafka-client@HADOOP.COM";
};
EOF
3、配置kafka-topics.sh等操作文件
复制 bin/kafka-topics.sh、bin/kafka-console-producer.sh、bin/kafka-console-consumer.sh 脚本，分别重命名为 bin/kafka-topics-sasl.sh、bin/kafka-console-producer-sasl.sh、bin/kafka-console-consumer-sasl.sh
$ cp $KAFKA_HOME/bin/kafka-topics.sh $KAFKA_HOME/bin/kafka-topics-sasl.sh $ cp $KAFKA_HOME/bin/kafka-console-producer.sh $KAFKA_HOME/bin/kafka-console-producer-sasl.sh $ cp $KAFKA_HOME/bin/kafka-console-consumer.sh $KAFKA_HOME/bin/kafka-console-consumer-sasl.sh $ cp $KAFKA_HOME/bin/kafka-consumer-groups.sh $KAFKA_HOME/bin/kafka-consumer-groups-sasl.sh
在倒数第二行增加如下配置（bin/kafka-xxx.sh）：
export KAFKA_OPTS="-Djava.security.krb5.conf=/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/kerberos/krb5.conf -Djava.security.auth.login.config=/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/kerberos/kafka-client-jaas.conf"
【温馨提示】上面创建了两个keytab文件，但是权限都是一样，都可以操作所有，所以在企业生成环境中，可以考虑把客户端的用户权限控制一下。
5）将配置文件copy到其它机器上
# kafka kerberos配置 $ scp -r $KAFKA_HOME/config/kerberos hadoop-node2:$KAFKA_HOME/config/ $ scp -r $KAFKA_HOME/config/kerberos hadoop-node3:$KAFKA_HOME/config/ # kafka server启动脚本 $ scp $KAFKA_HOME/bin/kafka-server-start-sasl.sh hadoop-node2:$KAFKA_HOME/bin/ $ scp $KAFKA_HOME/bin/kafka-server-start-sasl.sh hadoop-node3:$KAFKA_HOME/bin/ # 【温馨提示】注意修改broker.id、advertised.listeners # kafka server 配置 $ scp $KAFKA_HOME/config/server-sasl.properties hadoop-node2:$KAFKA_HOME/config/ $ scp $KAFKA_HOME/config/server-sasl.properties hadoop-node3:$KAFKA_HOME/config/ # topic操作 $ scp $KAFKA_HOME/bin/kafka-topics-sasl.sh hadoop-node2:$KAFKA_HOME/bin/ $ scp $KAFKA_HOME/bin/kafka-topics-sasl.sh hadoop-node3:$KAFKA_HOME/bin/ # 生产者 $ scp $KAFKA_HOME/bin/kafka-console-producer-sasl.sh hadoop-node2:$KAFKA_HOME/bin/ $ scp $KAFKA_HOME/bin/kafka-console-producer-sasl.sh hadoop-node3:$KAFKA_HOME/bin/ # 消费者 $ scp $KAFKA_HOME/bin/kafka-console-consumer-sasl.sh hadoop-node2:$KAFKA_HOME/bin/ $ scp $KAFKA_HOME/bin/kafka-console-consumer-sasl.sh hadoop-node3:$KAFKA_HOME/bin/ # 消费组 $ scp $KAFKA_HOME/bin/kafka-consumer-groups-sasl.sh hadoop-node2:$KAFKA_HOME/bin/ $ scp $KAFKA_HOME/bin/kafka-consumer-groups-sasl.sh hadoop-node3:$KAFKA_HOME/bin/
6）测试验证
1、查看topic 列表
运行 bin/kafka-server-start-sasl.sh 启动 Kafka 集群：
$ ./bin/kafka-server-start-sasl.sh config/server-sasl.properties # 还是跟之前的一样查询，会发现已经不能查询了 $ ./bin/kafka-topics-sasl.sh --list --bootstrap-server hadoop-node1:19092
服务端报的错误
客户端报的错误
正确访问方式
$ cd $KAFKA_HOME $ ./bin/kafka-topics-sasl.sh --list --bootstrap-server hadoop-node1:19092 --command-config config/kerberos/client.properties
上面是只启动了一台机器，现在验证ok了，就可以把三台kafka的机器都启动，要不然下面的实验会失败，因为之前kafka配置的就是集群模式。如果只启动一台，会报如下错误：
replication factor: 3 larger than available brokers: 1
【问题提示】需要修改
- 需要注意的是得把
kafka-server-node2.keytab和kafka-server-node3.keytab修改回kafka-server.keytab - 还得把
kafka-server-jaas.conf里面的主机名改成本机主机名 - 修改
config/server-sasl.properties文件里的broker.id和advertised.listeners
# hadoop-ndoe2 $ cd $KAFKA_HOME $ mv config/kerberos/kafka-server-node2.keytab config/kerberos/kafka-server.keytab $ vi config/kerberos/kafka-server-jaas.conf $ vi config/server-sasl.properties $ ./bin/kafka-server-start-sasl.sh -daemon config/server-sasl.properties # hadoop-ndoe2 $ cd $KAFKA_HOME $ mv config/kerberos/kafka-server-node3.keytab config/kerberos/kafka-server.keytab $ vi config/kerberos/kafka-server-jaas.conf $ vi config/server-sasl.properties $ ./bin/kafka-server-start-sasl.sh -daemon config/server-sasl.properties
2、创建topic
$ cd $KAFKA_HOME $ ./bin/kafka-topics-sasl.sh --create --topic test-sasl --bootstrap-server hadoop-node1:19092 --partitions 3 --replication-factor 3 --config retention.ms=259200000 --command-config config/kerberos/client.properties
3、查看topic详情
$ cd $KAFKA_HOME $ ./bin/kafka-topics-sasl.sh --bootstrap-server hadoop-node1:19092 --describe --topic test-sasl --command-config config/kerberos/client.properties
4、删除topic
$ cd $KAFKA_HOME $ ./bin/kafka-topics-sasl.sh --delete --topic test-sasl --bootstrap-server hadoop-node1:19092 --command-config config/kerberos/client.properties
4、生产/消费
运行 kafka-console-producer-sasl.sh 生产消息：
$ cd $KAFKA_HOME
$ ./bin/kafka-console-producer-sasl.sh --bootstrap-server hadoop-node1:19092 --topic test-sasl --producer.config config/kerberos/client.properties
{"id":"1","name":"n1","age":"20"}
{"id":"2","name":"n2","age":"21"}
{"id":"3","name":"n3","age":"22"}
运行 bin/kafka-console-consumer-sasl.sh 消费消息：
$ cd $KAFKA_HOME $ ./bin/kafka-console-consumer-sasl.sh --bootstrap-server hadoop-node1:19092 --topic test-sasl --from-beginning --consumer.config config/kerberos/client.properties #消费组 $ ./bin/kafka-console-consumer-sasl.sh --bootstrap-server hadoop-node1:19092 --topic test-sasl --from-beginning --group test-sasl --consumer.config config/kerberos/client.properties
5、查看消费组积压
$ cd $KAFKA_HOME $ ./bin/kafka-consumer-groups-sasl.sh --bootstrap-server hadoop-node1:19092 --describe --group test-sasl --command-config config/kerberos/client.properties
四、Kafka 用户密码认证（SASL/PLAIN ）
SASL/PLAIN 是一种简单的用户名/密码认证机制，通常与 TLS 一起用于加密以实现安全认证。Kafka 支持 SASL/PLAIN 的默认实现，可以按照此处所述进行扩展以供生产使用。
1）Kafka 服务端配置
1、配置jaas
在每个 Kafka 代理的配置目录中添加一个经过适当修改的 JAAS 文件，类似于下面的文件，我们将其命名为 kafka_server_jaas.conf 以用于本示例：
$ mkdir $KAFKA_HOME/config/userpwd
$ cat >$KAFKA_HOME/config/userpwd/kafka_server_jaas.conf <<EOF
KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="adminpasswd"
user_admin="adminpasswd"
user_kafka="123456"
user_producer="producerpwd"
user_consumer="consumerpwd";
};
EOF
- 该文件中username和password是定义zookeeper集群节点之间认证的用户名和密码
user_开头配置的用户kafka和密码kafka是提供给外部应用连接zookeeper使用的，后面kafka使用此用户连接zookeeper。上面配置的user_kafka="123456"，用户为kafka，密码为123456- username为admin的用户，和user为admin的用户，密码要保持一致，否则会认证失败
上述配置中，创建了三个用户，分别为admin、producer和consumer（创建多少个用户，可根据业务需要配置，用户名和密码可自定义设置）
2、配置服务启动参数（kafka-server-start.sh）
$ cd $KAFKA_HOME $ cp $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/bin/kafka-server-start-pwd.sh $ vi $KAFKA_HOME/bin/kafka-server-start-pwd.sh
在倒数第二行增加如下配置（bin/kafka-server-start-pwd.sh）：
export KAFKA_OPTS="-Dzookeeper.sasl.client=false -Dzookeeper.sasl.client.username=zk-server -Djava.security.auth.login.config=/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/userpwd/kafka_server_jaas.conf"
3、配置服务启动文件（server.properties）
$ cp $KAFKA_HOME/config/server.properties $KAFKA_HOME/config/server-pwd.properties
添加或修改如下配置：
listeners=SASL_PLAINTEXT://0.0.0.0:19092 # 不同的机器修改成对应的主机名 advertised.listeners=SASL_PLAINTEXT://hadoop-node1:19092 security.inter.broker.protocol=SASL_PLAINTEXT sasl.mechanism.inter.broker.protocol=PLAIN sasl.enabled.mechanisms=PLAIN authorizer.class.name=kafka.security.authorizer.AclAuthorizer allow.everyone.if.no.acl.found=true
2）Kafka 客户端配置
1、创建客户端认证配置文件
$ cat >$KAFKA_HOME/config/userpwd/kafka_client_jaas.conf <<EOF
KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="kafka"
password="123456";
};
EOF
2、在配置文件中添加属性
在 client.properties、producer.properties 、 consumer.properties 中为每个客户端配置 JAAS 配置属性。
- 新增
client.properties
$ cat >$KAFKA_HOME/config/userpwd/client.properties <<EOF security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN EOF
- 配置
producer.properties
$ cp $KAFKA_HOME/config/producer.properties $KAFKA_HOME/config/producer-pwd.properties # 在文件末尾添加 security.protocol=SASL_SSL sasl.mechanism=PLAIN
- 配置
consumer.properties
$ cp $KAFKA_HOME/config/consumer.properties $KAFKA_HOME/config/consumer-pwd.properties # 在文件末尾添加 security.protocol=SASL_PLAINTEXT sasl.mechanism=PLAIN
3、配置kafka-topics.sh等操作文件
$ cp $KAFKA_HOME/bin/kafka-topics.sh $KAFKA_HOME/bin/kafka-topics-pwd.sh $ cp $KAFKA_HOME/bin/kafka-console-producer.sh $KAFKA_HOME/bin/kafka-console-producer-pwd.sh $ cp $KAFKA_HOME/bin/kafka-console-consumer.sh $KAFKA_HOME/bin/kafka-console-consumer-pwd.sh $ cp $KAFKA_HOME/bin/kafka-consumer-groups.sh $KAFKA_HOME/bin/kafka-consumer-groups-pwd.sh
在倒数第二行增加如下配置（bin/kafka-xxx.sh）：
export KAFKA_OPTS="-Djava.security.auth.login.config=/opt/bigdata/hadoop/server/kafka_2.13-3.1.1/config/userpwd/kafka_client_jaas.conf"
3）测试验证
启动服务
$ cd $KAFKA_HOME $ ./bin/kafka-server-start-pwd.sh config/server-pwd.properties
1、查看topic列表
$ cd $KAFKA_HOME $ ./bin/kafka-topics-pwd.sh --list --bootstrap-server hadoop-node1:19092 --command-config config/userpwd/client.properties
上面已经在单机上验证是没问题了，现在就启动三台机器在验证
# 首先将修改的文件copy到另外两台机器上 # kafka 账号密码认证配置 scp -r $KAFKA_HOME/config/userpwd hadoop-node2:$KAFKA_HOME/config/ scp -r $KAFKA_HOME/config/userpwd hadoop-node3:$KAFKA_HOME/config/ # kafka server启动脚本 scp $KAFKA_HOME/bin/kafka-server-start-pwd.sh hadoop-node2:$KAFKA_HOME/bin/ scp $KAFKA_HOME/bin/kafka-server-start-pwd.sh hadoop-node3:$KAFKA_HOME/bin/ # 【温馨提示】注意修改broker.id、advertised.listeners # kafka server 配置 scp $KAFKA_HOME/config/server-pwd.properties hadoop-node2:$KAFKA_HOME/config/ scp $KAFKA_HOME/config/server-pwd.properties hadoop-node3:$KAFKA_HOME/config/ # topic scp $KAFKA_HOME/bin/kafka-topics-pwd.sh hadoop-node2:$KAFKA_HOME/bin/ scp $KAFKA_HOME/bin/kafka-topics-pwd.sh hadoop-node3:$KAFKA_HOME/bin/ # 生成者 scp $KAFKA_HOME/bin/kafka-console-producer-pwd.sh hadoop-node2:$KAFKA_HOME/bin/ scp $KAFKA_HOME/bin/kafka-console-producer-pwd.sh hadoop-node3:$KAFKA_HOME/bin/ # 消费者 scp $KAFKA_HOME/bin/kafka-console-consumer-pwd.sh hadoop-node2:$KAFKA_HOME/bin/ scp $KAFKA_HOME/bin/kafka-console-consumer-pwd.sh hadoop-node3:$KAFKA_HOME/bin/ # 消费组 scp $KAFKA_HOME/bin/kafka-consumer-groups-pwd.sh hadoop-node2:$KAFKA_HOME/bin/ scp $KAFKA_HOME/bin/kafka-consumer-groups-pwd.sh hadoop-node3:$KAFKA_HOME/bin/
启动服务
$ cd $KAFKA_HOME $ ./bin/kafka-server-start-pwd.sh -daemon config/server-pwd.properties
2、创建topic
$ cd $KAFKA_HOME $ ./bin/kafka-topics-pwd.sh --create --topic test-pwd --bootstrap-server hadoop-node1:19092 --partitions 3 --replication-factor 3 --config retention.ms=259200000 --command-config config/userpwd/client.properties
3、查看topic详情
$ cd $KAFKA_HOME $ ./bin/kafka-topics-pwd.sh --bootstrap-server hadoop-node1:19092 --describe --topic test-pwd --command-config config/userpwd/client.properties
4、删除topic
$ cd $KAFKA_HOME $ ./bin/kafka-topics-pwd.sh --delete --topic test-pwd --bootstrap-server hadoop-node1:19092 --command-config config/userpwd/client.properties
4、生产/消费
运行 kafka-console-producer-sasl.sh 生产消息：
$ cd $KAFKA_HOME
$ ./bin/kafka-console-producer-pwd.sh --bootstrap-server hadoop-node1:19092 --topic test-pwd --producer.config config/userpwd/client.properties
{"id":"1","name":"n1","age":"20"}
{"id":"2","name":"n2","age":"21"}
{"id":"3","name":"n3","age":"22"}
运行 bin/kafka-console-consumer-sasl.sh 消费消息：
$ cd $KAFKA_HOME $ ./bin/kafka-console-consumer-pwd.sh --bootstrap-server hadoop-node1:19092 --topic test-pwd --from-beginning --consumer.config config/userpwd/client.properties #消费组 $ ./bin/kafka-console-consumer-pwd.sh --bootstrap-server hadoop-node1:19092 --topic test-pwd --from-beginning --group test-sasl --consumer.config config/userpwd/client.properties
5、查看消费组积压
$ cd $KAFKA_HOME $ ./bin/kafka-consumer-groups-pwd.sh --bootstrap-server hadoop-node1:19092 --describe --group test-pwd --command-config config/userpwd/client.properties
五、CDH 启用Kerberos认证
CDH的安装可以参考我之前的文章：大数据Hadoop之——Cloudera Hadoop（CM 6.3.1+CDH 6.3.2环境部署）
1）安装Kerberos
Kerberos详解安装讲解可以参考我之前的文章：Kerberos认证原理与环境部署
2）为CM创建管理员主体/实例
$ kadmin.local -q "add_principal cloudera-scm/admin" 输入密码：123456
3）在CM上开启Kerberos认证
全选
通过CM管理krb5.conf，这里我已经提前配置好了，所以不选了。直接下一步继续
配置CM的kerberos管理用户（cloudera-scm/admin），上面创建的
自动导入导入 KDC Account Manager 凭据 ，直接下一步继续
已经装好了kerberos客户端，直接下一步继续即可
自动停止集群为每个组件创建一个kerberos认证用户，然后再自动启动每个组件，等待一段时间即可
完成kerberos安装与配置
查看CM自动创建的kerberos用户
$ kadmin.local -q "listprincs"
3）简单使用
用户访问服务认证
开启Kerberos安全认证之后，日常的访问服务（例如访问HDFS，消费Kafka topic等）都需要先进行安全认证
1、在Kerberos数据库中创建用户主体/实例
$ kadmin.local -q "addprinc hive/hive@HADOOP.COM" 输入密码：123456 再确认密码：123456
2、进行用户认证访问HDFS
# 清理当前用户 $ kdestroy $ hadoop fs -ls / # 发现如果没登录，就访问不了了
登录
# 生成ketab文件，一般常用keytab文件去做认证 $ kadmin.local -q "xst -k /root/hive.keytab hive/hive@HADOOP.COM" $ kinit -kt /root/hive.keytab hive/hive $ klist $ hadoop fs -ls /
访问其它组件也类似，这里就不再演示了，感兴趣的小伙伴可以去练习验证一下，本章也到这先结束了，有疑问的小伙伴，欢迎给我留言哦，后续会分享更多关于大数据的文章，请小伙伴耐心等待~
这篇关于大数据Hadoop之——Kafka鉴权认证（Kafka kerberos认证+kafka账号密码认证+CDH Kerberos认证）的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！
您可能喜欢
-
Fetch / Axios学习：入门教程与实战指南11-14
-
Typescript 类型课程入门教程11-14
-
Fetch / Axios课程：初学者必看的网络请求教程11-14
-
Styled-components课程：初学者指南11-14
-
pre-commit 自动化测试课程：入门教程与实践指南11-13
-
什么是AIGC？如何使用AIGC技术辅助办公?11-13
-
Slicm 框架怎么进行用户认证？-icode9专业技术文章分享11-13
-
在查询时将 map_coord 列的值转换为字符串有哪些方法？-icode9专业技术文章分享11-13
-
如何将微信地区改成自定义文案？-icode9专业技术文章分享11-13
-
DNS 缓存存在问题有哪些症状和解决方法？-icode9专业技术文章分享11-13
-
HTTP 状态码（405）-Method Not Allowed是什么意思？-icode9专业技术文章分享11-13
-
HTTP 状态码（500）-Internal Server Error是什么意思？-icode9专业技术文章分享11-13
-
在 Element UI 中无法修改 $confirm 的取消按钮文字是什么原因？-icode9专业技术文章分享11-13
-
unity XR是什么？-icode9专业技术文章分享11-13
-
伴随矩阵是什么？-icode9专业技术文章分享11-13
栏目导航
