目录
1. Kafka配置Kerberos
2. 客户端操作Kafka
3. Java API操作Kafka
4. StructuredStreaming操作Kafka
5. Flink 操作Kafka
技术连载系列，前面内容请参考前面连载11内容:Kerberos安全认证-连载11-HBase Kerberos安全配置及访问_IT贫道的博客-CSDN博客
1. Kafka配置Kerberos
Kafka也支持通过Kerberos进行认证，避免非法用户操作读取Kafka中的数据，对Kafka进行Kerberos认证可以按照如下步骤实现。
1) 创建Kafka服务Princial主体并写入到keytab文件
在kerberos服务端node1节点执行如下命令创建Kafka服务主体:
[root@node1 ~]# kadmin.local -q "addprinc -pw 123456 kafka/node1"
[root@node1 ~]# kadmin.local -q "addprinc -pw 123456 kafka/node2"
[root@node1 ~]# kadmin.local -q "addprinc -pw 123456 kafka/node3"
在kerberos服务端node1节点执行如下命令将Kafka服务主体写入到keytab文件。
#node1节点执行命令，将主体写入到keytab
[root@node1 ~]# kadmin.local -q "ktadd -norandkey -kt /home/keytabs/kafka.service.keytab kafka/node1@EXAMPLE.COM"
[root@node1 ~]# kadmin.local -q "ktadd -norandkey -kt /home/keytabs/kafka.service.keytab kafka/node2@EXAMPLE.COM"
[root@node1 ~]# kadmin.local -q "ktadd -norandkey -kt /home/keytabs/kafka.service.keytab kafka/node3@EXAMPLE.COM"
以上命令执行完成后，在node1节点/home/keytabs目录下生成kafka.service.keytab文件，将该文件分发到各个节点并赋权，这里可以只发送到node1~node3 Kafka所在节点，为了保证各个大数据集群节点的keytabs一致，这里分发到所有节点。
[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node2:/home/keytabs/
[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node3:/home/keytabs/
[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node4:/home/keytabs/
[root@node1 ~]# scp /home/keytabs/kafka.service.keytab node5:/home/keytabs/
分发完成后，在集群各个节点上执行如下命令，修改kafka.service.keytab密钥文件访问权限：
chmod 770 /home/keytabs/kafka.service.keytab
2) 修改配置server.properties文件
在Kafka各个节点KAFKA_HOME/config/server.properties文件中加入如下配置以支持Kerberos安全认证。
#在node1~node3所有节点单独配置
listeners=SASL_PLAINTEXT://:9092
inter.broker.listener.name=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=GSSAPI
sasl.enabled.mechanisms=GSSAPI
sasl.kerberos.service.name=kafka
authorizer.class.name=kafka.security.authorizer.AclAuthorizer
zookeeper.set.acl=false
allow.everyone.if.no.acl.found=true
该配置需要在node1~node3所有节点配置，以上参数解释如下：
- listeners=SASL_PLAINTEXT://:9092
指定kafka监听的协议和端口，SASL_PLAINTEXT表示使用SASL（Simple Authentication and Security Layer）机制进行认证和加密通信。
- inter.broker.listener.name=SASL_PLAINTEXT
指定Kafka Broker 之间通信使用SASL_PLAINTEXT机制进行认证和加密通信。
- sasl.mechanism.inter.broker.protocol=GSSAPI
指定broker之间通信使用SASL机制，GSSAPI是一种基于Kerberos的SASL机制，它使用GSS-API（Generic Security Services Application Programming Interface）进行认证。
- sasl.enabled.mechanisms=GSSAPI
只启用了GSSAPI机制，表示Kafka只接受使用Kerberos进行认证的连接。
- sasl.kerberos.service.name=kafka
指定Kafka在Kerberos中注册的服务名称，以便用于进行身份验证和授权。
- authorizer.class.name=kafka.security.authorizer.AclAuthorizer
指定Kafka使用的授权器类。
是否在ZooKeeper中设置ACL（Access Control List），这里设置为false，表示不对ZooKeeper节点设置ACL。
- allow.everyone.if.no.acl.found=true
指定当没有匹配的ACL规则时，是否允许所有用户访问。
3) 准备kafka_jaas.conf文件
在node1~node3各个节点中准备kafka_jaas.conf文件，该文件配置kafka服务端和zookeeper客户端的身份验证和授权配置，由于zookeeper开启了Kerberos认证，所以这里需要进行zookeeper客户端的身份验证配置。
这里在各个kafka节点KAFKA_HOME/config/目录中创建kafka_jaas.conf文件，内容如下:
KafkaServer {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/home/keytabs/kafka.service.keytab"
serviceName="kafka"
principal="kafka/node1@EXAMPLE.COM";
};
Client {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
serviceName="zookeeper"
keyTab="/home/keytabs/zookeeper.service.keytab" principal="zookeeper/node3@EXAMPLE.COM";
};
以上在可以在node1 kafka客户端配置完成后，分发到node2~node3节点中，如下：
#在node1节点分发，分发后在node2、node3节点配置对应Server principal
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_jaas.conf node2:/software/kafka_2.12-3.3.1/config/
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_jaas.conf node3:/software/kafka_2.12-3.3.1/config/
注意：在node1~node3各个kafka节点中该文件中的KafkaServer对应的principal不同，为对应各个节点hostname。
4) 准备kafka_client_jaas.conf
在各个kafka节点配置kafka_client_jaas.conf配置文件，该文件作用主要是对kafka 客户端进行身份认证。这里在node1~node3节点KAFKA_HOME/config/中创建kafka_client_jaas.conf文件，内容如下：
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="/home/keytabs/kafka.service.keytab"
serviceName="kafka"
principal="kafka/node1@EXAMPLE.COM";
};
可以在node1节点配置该文件后分发到node2~node3节点中：
#分发到node2、node3节点，需要在对应节点配置对应的principal
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf node2:/software/kafka_2.12-3.3.1/config/
[root@node1 config]# scp /software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf node3:/software/kafka_2.12-3.3.1/config/
以上文件分发完成后，需要在对应节点修改配置对应的Principal信息为对应的hostname。
5) 修改启动脚本kafka-server-start.sh
在kafka各个节点中配置KAFKA_HOME/bin/kafka-server-start.sh启动脚本，在该脚本中加入kafka_jaas.conf配置，加入的内容如下：
#在node1~node3各个节点都要配置kafka-server-start.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_jaas.conf"
6) 修改kafka操作脚本
在当前集群中node1~node3节点是kafka服务端，同时如果在这3个节点上进行kafka 命令操作，这三个节点也是kafka客户端。在操作kafka时我们通常会操作KAFKA_HOME中的kafka-topic.sh、kafka-console-producer.sh、kafka-console-consumer.sh脚本，这些脚本需要进行kerberos认证，可以通过前面配置的kafka_client_jaas.conf文件进行Kerberos认证，所以这里在各个脚本中加入如下配置，避免在操作对应脚本时没有进行认证从而没有操作权限。
在node1~node3各个kafka 客户端配置以上脚本，可以先在node1节点进行配置各文件然后分发到其他kafka客户端，对应操作文件增加如下配置:
#vim /software/kafka_2.12-3.3.1/bin/kafka-topics.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf"
#vim /software/kafka_2.12-3.3.1/bin/kafka-console-producer.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf"
#vim /software/kafka_2.12-3.3.1/bin/kafka-console-consumer.sh
export KAFKA_OPTS="-Djava.security.krb5.conf=/etc/krb5.conf -Djava.security.auth.login.config=/software/kafka_2.12-3.3.1/config/kafka_client_jaas.conf"
在node1 kafka 客户端配置完成后，将配置后的各个脚本文件分发到其他kafka客户端节点上：
[root@node1 ~]# cd /software/kafka_2.12-3.3.1/bin/
[root@node1 bin]# scp ./kafka-topics.sh ./kafka-console-producer.sh ./kafka-console-consumer.sh node2:`pwd`
[root@node1 bin]# scp ./kafka-topics.sh ./kafka-console-producer.sh ./kafka-console-consumer.sh node3:`pwd`
7) 准备client.properites
在node1~node3各个kafka 客户端中准备client.properties配置文件，该配置文件内容如下，这里将该文件创建在各个节点的/root目录下。
security.protocol=SASL_PLAINTEXT
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka
当Kafka通过Kerberos认证后，在执行KAFKA_HOME/bin目录下的脚本时，需要使用正确的协议、SASL机制及kerberos服务，所以这里将以上信息配置到client.properties文件中，在执行各个脚本时需要通过参数指定该文件，这样客户端可以和服务端正常通信。
在node1配置完成/root/client.properties文件后，分发到node2~node3节点中：
[root@node1 ~]# scp ./client.properties node2:`pwd`
[root@node1 ~]# scp ./client.properties node3:`pwd`
8) 启动kafka集群
启动kafka集群前需要先启动Zookeeper，然后在各个kafka服务节点启动kafka，完成kafka集群启动。操作如下：
#node3~node5各节点启动zookeeper
zkServer.sh start
#node1~node3各节点启动kafka
startKafka.sh
2. 客户端操作Kafka
启动Kafka集群后，通过如下命令在Kafka集群查询、创建topic以及向topic中写入数据，以下命令可以执行在node1~node3各个kafka 客户端中。
#创建kafka topic
kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --create --topic test --partitions 3 --replication-factor 1 --command-config /root/client.properties
#查看集群topic
kafka-topics.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --list --command-config /root/client.properties
#向kafka topic中写入数据
[root@node1 ~]# kafka-console-producer.sh --broker-list node1:9092,node2:9092,node3:9092 --topic test --producer.config /root/client.properties
>1
>2
>3
>4
>5
#读取kafka topic中的数据
[root@node1 ~]# kafka-console-consumer.sh --bootstrap-server node1:9092,node2:9092,node3:9092 --topic test --from-beginning --consumer.config /root/client.properties
1
2
3
4
5
3. Java API操作Kafka
可以按照如下步骤实现Java API操作Kerberos认证的Kafka数据。
1) 准备krb5.conf文件
将node1 kerberos服务端/etc/krb5.conf文件存放在IDEA项目中的resources资源目录中或者本地Window固定的某个目录中，用于编写代码时指定访问Kerberos的Realm。
2) 准备用户keytab文件
在kerberos服务端node1节点上将生成的kafka.server.keytab文件存入到window路径中，这里放在项目resource资源目录下，后续需要该文件进行客户端认证。
3) 准备kafka_client_jaas.conf文件
将Kafka 中kafka_client_jaas.conf文件放在window中某个路径中，并修改该文件中keytab路径为window路径：
KafkaClient {
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
storeKey=true
keyTab="D:/idea_space/KerberosAuth/KerberosAuthKafka/src/main/resources/kafka.service.keytab"
serviceName="kafka"
principal="kafka/node1@EXAMPLE.COM";
};
特别需要注意的是该文件中指定window路径时使用“/”或“\\”隔开各目录，否则客户端认证时读取不到keytab文件。这里将修改后的kafka_client_jaas.conf文件存入到项目resource资源目录下。
4) 编写Java代码向Kafka topic中读写数据
编写代码前，需要在项目pom.xml中引入如下依赖：
<!-- kafka client依赖包 -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>3.3.1</version>
</dependency>
Java API读写Kafka Topic代码如下：
/**
* Java API 操作Kerbros认证的Kafka
* 使用 JAAS 来进行 Kerberos 认证
* 注意：kafka_client_jaas.conf文件中的keytab文件路径需要使用双斜杠或者反单斜杠
*/
public class OperateAuthKafka {
public static void main(String[] args) {
//准备JAAS配置文件路径
String kafkaClientJaasFile = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\kafka_client_jaas.conf";
// Kerberos配置文件路径
String krb5FilePath = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\krb5.conf";
System.setProperty("java.security.auth.login.config", kafkaClientJaasFile);
System.setProperty("java.security.krb5.conf", krb5FilePath);
Properties props = new Properties();
props.setProperty("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
props.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
//kerberos安全认证
props.setProperty("security.protocol", "SASL_PLAINTEXT");
props.setProperty("sasl.mechanism", "GSSAPI");
props.setProperty("sasl.kerberos.service.name", "kafka");
//向Kafka topic中发送消息
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
kafkaProducer.send(new ProducerRecord<>("test", "100"));
kafkaProducer.send(new ProducerRecord<>("test", "200"));
kafkaProducer.send(new ProducerRecord<>("test", "300"));
kafkaProducer.send(new ProducerRecord<>("test", "400"));
kafkaProducer.send(new ProducerRecord<>("test", "500"));
kafkaProducer.close();
System.out.println("消息发送成功");
/**
* 从Kafka topic中消费消息
*/
props.setProperty("group.id", "test"+ UUID.randomUUID());
//设置消费的位置，earliest表示从头开始消费，latest表示从最新的位置开始消费
props.setProperty("auto.offset.reset", "earliest");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
kafkaConsumer.subscribe(Arrays.asList("test"));
while (true) {
// 拉取数据
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : consumerRecords) {
// 获取数据对应的分区号
int partition = record.partition();
// 对应数据值
String value = record.value();
//对应数据的偏移量
long lastoffset = record.offset();
//对应数据发送的key
String key = record.key();
System.out.println("数据的key为:"+ key +
",数据的value为:" + value +
",数据的offset为:"+ lastoffset +
",数据的分区为:"+ partition);
}
}
}
}
4. StructuredStreaming操作Kafka
StructuredStreaming操作Kafka时同样需要准备krb5.conf、kafka.service.keytab、kafka_client_jaas.conf配置文件，步骤参考Java API操作Kafka部分。
编写代码前，需要在项目pom.xml中引入如下依赖：
<!-- SparkSQL -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>3.4.0</version>
</dependency>
<!-- Kafka 0.10+ Source For Structured Streaming-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql-kafka-0-10_2.12</artifactId>
<version>3.4.0</version>
</dependency>
StructuredStreaming操作Kafka代码如下：
/**
* StructuredStreaming 读取Kerberos 认证的Kafka数据
*/
public class StructuredStreamingReadAuthKafka {
public static void main(String[] args) throws TimeoutException, StreamingQueryException {
//准备JAAS配置文件路径
String kafkaClientJaasFile = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\kafka_client_jaas.conf";
// Kerberos配置文件路径
String krb5FilePath = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\krb5.conf";
System.setProperty("java.security.auth.login.config", kafkaClientJaasFile);
System.setProperty("java.security.krb5.conf", krb5FilePath);
//1.创建对象
SparkSession spark = SparkSession.builder()
.master("local")
.appName("kafka source")
.config("spark.sql.shuffle.partitions", 1)
.getOrCreate();
spark.sparkContext().setLogLevel("Error");
//2.读取kafka 数据
Dataset<Row> df = spark.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "node1:9092,node2:9092,node3:9092")
.option("subscribe", "test")
.option("startingOffsets", "earliest")
//kerberos安全认证
.option("kafka.security.protocol", "SASL_PLAINTEXT")
.option("kafka.sasl.mechanism", "GSSAPI")
.option("kafka.sasl.kerberos.service.name", "kafka")
.load();
Dataset<Row> result = df.selectExpr("cast (key as string)", "cast (value as string)");
StreamingQuery query = result.writeStream()
.format("console")
.start();
query.awaitTermination();
}
}
5. Flink 操作Kafka
Flink操作Kafka时同样需要准备krb5.conf、kafka.service.keytab、kafka_client_jaas.conf配置文件，步骤参考Java API操作Kafka部分。
编写代码前，需要在项目pom.xml中引入如下依赖：
<!-- Flink批和流开发依赖包 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients</artifactId>
<version>1.16.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java</artifactId>
<version>1.16.0</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-kafka</artifactId>
<version>1.16.0</version>
</dependency>
Flink操作Kafka代码如下：
/**
* Flink 读取Kerberos 认证的Kafka数据
*/
public class FlinkReadAuthKafka {
public static void main(String[] args) throws Exception {
//准备JAAS配置文件路径
String kafkaClientJaasFile = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\kafka_client_jaas.conf";
// Kerberos配置文件路径
String krb5FilePath = "D:\\idea_space\\KerberosAuth\\KerberosAuthKafka\\src\\main\\resources\\krb5.conf";
System.setProperty("java.security.auth.login.config", kafkaClientJaasFile);
System.setProperty("java.security.krb5.conf", krb5FilePath);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
KafkaSource<Tuple2<String, String>> kafkaSource = KafkaSource.<Tuple2<String, String>>builder()
.setBootstrapServers("node1:9092,node2:9092,node3:9092") //设置Kafka 集群节点
.setTopics("test") //设置读取的topic
.setGroupId("my-test-group") //设置消费者组
//kerberos安全认证
.setProperty("security.protocol", "SASL_PLAINTEXT")
.setProperty("sasl.mechanism", "GSSAPI")
.setProperty("sasl.kerberos.service.name", "kafka")
.setStartingOffsets(OffsetsInitializer.earliest()) //设置读取数据位置
.setDeserializer(new KafkaRecordDeserializationSchema<Tuple2<String, String>>() {
//设置key ,value 数据获取后如何处理
@Override
public void deserialize(ConsumerRecord<byte[], byte[]> consumerRecord, Collector<Tuple2<String, String>> collector) throws IOException {
String key = null;
String value = null;
if(consumerRecord.key() != null){
key = new String(consumerRecord.key(), "UTF-8");
}
if(consumerRecord.value() != null){
value = new String(consumerRecord.value(), "UTF-8");
}
collector.collect(Tuple2.of(key, value));
}
//设置置返回的二元组类型
@Override
public TypeInformation<Tuple2<String, String>> getProducedType() {
return TypeInformation.of(new TypeHint<Tuple2<String, String>>() {
});
}
})
.build();
DataStreamSource<Tuple2<String, String>> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka-source");
kafkaDS.print();
env.execute();
}
}
欢迎点赞、评论、收藏，关注IT贫道，获取IT技术知识！
