如何为Kafka加上账号密码（二）

认证策略SASL/PLAIN

上篇文章中我们讲解了Kafka认证方式和基础概念，并比较了不同方式的使用场景。

我们在《2024年了，如何更好的搭建Kafka集群？》中集群统一使用PLAINTEXT通信。Kafka通常是在内网使用，但也有特殊的使用场景需要暴漏到公网上，如果未设置认证的Kafka集群允许通过公网访问，或暴漏给全部研发人员是极不安全的方式。

本小节我们就为Kafka添加最简单的认证方式，也就是SASL_PLAINTEXT（即SASL/PLAIN+ 非加密通道）。

配置服务集群节点

Kafka有三个地方可以做认证：
borker节点之间的认证、controller节点间的认证、外部客户端连接集群认证。公司内部使用我们只需要配置外部客户端连接时认证就可以了。本文基于kraft单节点部署Kafka，该节点既是controller又是broker节点。

在开始前，请先前往官网下载安装包（本文使用 Kafka-v3.6.1）：
https://kafka.apache.org/downloads

1. 创建jaas配置文件

文件：volume/config/kafka-server-jaas.conf

KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requireduser_admin="123456";
};

账号密码以user_{USERNAME}="PASSWORD"这种形式配置，固定写死在jaas的配置文件中。

2. 修改配置文件

修改config/kraft/server.properties如下：

# 修改listeners和advertised
listeners=PLAINTEXT://:9092,CONTROLLER://:9093,SASL_PLAINTEXT://:19092
advertised.listeners=PLAINTEXT://192.168.56.103:9092,SASL_PLAINTEXT://192.168.56.103:19092
# 添加mechanisms配置
sasl.enabled.mechanisms=PLAIN

虽然该实验我们只有一个节点，但我们按照集群的方式分配端口：

9092：用于broker节点内部通讯和内网客户端通信
19092：用于broker节点外部通讯和外网客户端通信
9093：用于多个controller节点之间的通信

我们只需要对外部流量做认证即可，也就是所有通过19092端口的通信需要经过认证。

配置文件中有三处需要注意：

sasl.enabled.mechanisms=PLAIN 意思是采用PLAIN这种方式做认证。
listeners中的SASL_PLAINTEXT监听器，表示监听19092端口，并对经过该端口的通信做认证。
advertised.listeners中的SASL_PLAINTEXT监听器，表示仅允许符合192.168.56.103:19092的流量包抵达SASL_PLAINTEXT监听器。

关于advertised.listeners，Kafka有一套复杂的监听器系统。Kafka允许定义多个监听器监听不同端口，将不同的流量划分到不同的端口。这样的好处是当一个端口流量较大时，不影响其它端口的通信。

3. 启动节点

执行下面这条命令启动节点：

KAFKA_OPTS=-Djava.security.auth.login.config=config/kraft/jaas.conf  bin/kafka-server-start.sh config/kraft/server.properties

环境变量KAFKA_OPTS用于指定JAAS配置文件。

经过上面3个步骤服务端就配置好了，下面我们使用客户端验证认证是否生效。

配置客户端

我们将介绍三种客户端连接方式：

kafka命令工具测试
offset explorer图形工具连接kafka
flink连接kafka

kafka命令工具测试

在config目录中创建admin.conf

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="123456";

提示：确保账号密码与服务端配置一致


# 创建topic
bin/kafka-topics.sh --create --topic test-topic --bootstrap-server=192.168.56.103:19092 --command-config=config/admin.conf
# 查看topic
bin/kafka-topics.sh --list --bootstrap-server=192.168.56.103:19092 --command-config=config/admin.conf

使用图形工具Offset explorer连接kafka

![[attach/Pasted image 20240204135019.png]]
此处只需要填写好集群名称即可，我们没有用zk版，zookeeper相关配置无需修改。
![[attach/Pasted image 20240127151018.png]]
![[attach/Pasted image 20240127151059.png]]
![[attach/Pasted image 20240127151109.png]]

这三处设置好，就可以正常连接了。

Flink连接kafka

在无认证的基础上，额外添加三个认证参数：

SECURITY_PROTOCOL_CONFIG
SASL_MECHANISM
SASL_JAAS_CONFIG

        Properties properties = new Properties();properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");properties.setProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5");properties.setProperty(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");properties.setProperty(SaslConfigs.SASL_MECHANISM, "PLAIN");properties.setProperty(SaslConfigs.SASL_JAAS_CONFIG, "org.apache.kafka.common.security.plain.PlainLoginModule required username='admin' password='123456';");/*org.apache.flink.streaming包下面的消费者和生产者已被标记不建议使用org.apache.flink.connector.kafka 推荐使用该包中的工具类*/KafkaSource<String> kafkaSource = KafkaSource.<String>builder().setBootstrapServers("192.168.56.103:19092").setTopics("test-topic").setGroupId("test").setValueOnlyDeserializer(new SimpleStringSchema()).setStartingOffsets(OffsetsInitializer.earliest()).setProperties(properties).build();final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.enableCheckpointing(10000);env.setParallelism(1);DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "test-tip");stream.print();env.execute("Kafka to Print");