VB.net 2010 视频教程 VB.net 2010 视频教程 python基础视频教程
SQL Server 2008 视频教程 c#入门经典教程 Visual Basic从门到精通视频教程
当前位置:
首页 > 编程开发 > Java教程 >
  • Kafka 入门实战(4)--开启 Kerberos 认证

本主要介绍在 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);
    }
}
复制代码

 原文:https://www.cnblogs.com/wuyongyin/p/15668848.html


相关教程