KRaft使用SASL_PLAINTEXT进行认证

2023-12-03 23:20
文章标签 进行 使用 认证 kraft sasl plaintext

本文主要是介绍KRaft使用SASL_PLAINTEXT进行认证，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

需要有KRaft相关的基础，才行。可参阅之前学习记录Kafka

一、配置

首先需要了解SASL的含义，SASL全称为Simple Authentication and Security Layer，它主要是用于在客户端和服务器之间提供安全的身份验证机制。

Kafka 支持以下几种 SASL 验证机制如下

  • GSSAPI (Kerberos)
  • PLAIN
  • SCRAM-SHA-256
  • SCRAM-SHA-512
  • OAUTHBEARER

其中，PLAIN相对来说更简单，本文就记录SASL/PLAIN的配置与使用。

1.1 配置单机SASL/PLAIN认证

安装的过程参考一键安装的脚本，安装好Kafka后，按照如下操作进行配置

1.) 创建kafka_server_jaas.conf

KafkaServer {org.apache.kafka.common.security.plain.PlainLoginModule requiredusername="admin"password="admin-secret"user_admin="admin-secret"user_alice="alice-secret";
};

user_admin="admin-secret"表示用户名admin，对应的密码为admin-secret。user_alice同理。

username和password表示节点建立集群时，需要验证的身份信息，只有验证通过的节点，方能成功建立集群。

2.)进入到kafka的bin路径中， 复制一份kafka-server-start.sh出来。

cp kafka-server-start.sh kafka-server-start-jaas.sh

并修改复制出来的文件，添加三行。

if [ "x$KAFKA_OPTS" ]; thenexport KAFKA_OPTS="-Djava.security.auth.login.config=/root/kafka_server_jaas.conf"
fi

3.) 将config/kraft/server.properties再复制一份。

cp server.properties server.jaas.properties

并修改复制出来的文件，修改参数如下。其中advertised.listeners中的ip要改为实际使用的ip地址。

sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAINlisteners=SASL_PLAINTEXT://:9092,CONTROLLER://:9093
inter.broker.listener.name=SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://10.0.0.10:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

其中，listener.security.protocol.map的格式为{别名}:{监听器类型}，配置了别名，那么在配置listeners时，就可以通过{别名}://:{端口}来进行配置。前缀为 SASL 的两个参数是新增的，目的是为了配置 SASL 认证，其表示的含义如下

  • sasl.enabled.mechanisms：指定Kafka进行SASL验证的机制。PLAIN表示使用文本验证。
  • sasl.mechanism.inter.broker.protocol：指定broker之间通信所使用的SASL验证的机制。PLAIN表示使用文本验证。

Kafka支持不同的监听器类型，如下

监听器类型数据加密身份验证适用场景
PLAINTEXT————内网环境、安全性有保障
SSL/TLSSSL/TLS——在公网传输敏感数据
SASL_PLAINTEXT——SASL需要身份验证但不需要数据加密的环境
SASL_SSLSSL/TLSSASL安全性要求极高的环境

4.) 启动

格式化集群信息，并启动

/root/kafka_2.13-3.3.1/bin/kafka-storage.sh format -t T1CYXg2DQPmdSYSUI-FNFw -c /root/kafka_2.13-3.3.1/config/kraft/server.jaas.properties
/root/kafka_2.13-3.3.1/bin/kafka-server-start-jaas.sh /root/kafka_2.13-3.3.1/config/kraft/server.jaas.properties

会了单节点的配置，集群的配置只是简单调下参数，不过还是记录一下。

具体的配置参阅1.2即可。

1.2 配置集群SASL/PLAIN认证

准备三台机器

  • 10.0.0.101
  • 10.0.0.102
  • 10.0.0.102

按照单机Kafka的配置步骤进行配置，但是在第三步时，略有调整。

10.0.0.101的示例配置文件如下。每个节点中，需要单独修改三个参数，分别为node.id、controller.quorum.voters、advertised.listeners。

process.roles=broker,controller
node.id=1
controller.quorum.voters=1@10.0.0.101:9093,2@10.0.0.102:9093,3@10.0.0.103:9093
listeners=SASL_PLAINTEXT://:9092,CONTROLLER://:9093
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
inter.broker.listener.name=SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://10.0.0.101:9092
controller.listener.names=CONTROLLER
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL

其中node.id表示节点的编号，controller.quorum.voters表示集群中的所有具有controller角色的节点，其配置格式为{node.id}@{ip}:{port}。

在三节点上均执行如下命令

/root/kafka_2.13-3.3.1/bin/kafka-storage.sh format -t T1CYXg2DQPmdSYSUI-FNFw -c /root/kafka_2.13-3.3.1/config/kraft/server.jaas.properties
/root/kafka_2.13-3.3.1/bin/kafka-server-start-jaas.sh /root/kafka_2.13-3.3.1/config/kraft/server.jaas.properties

启动后日志如图。

1.3 client接入授权

1.) 如果kafka未开启身份认证

./kafka-consumer-groups.sh --bootstrap-server 10.0.0.101:9092 --describe --all-groups

2.) 如果kafka开启了身份认证，则需要创建存储身份认证的文件kafka.properties

security.protocol: SASL_PLAINTEXT
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";

之后再执行命令

./kafka-consumer-groups.sh --bootstrap-server 10.0.0.101:9092 --describe --all-groups --command-config ./kafka.properties

1.4 集群为何是奇数

常见的分布式一致性算法

  1. Paxos
  2. Raft

KRaft全称为Kafka Raft，是基于Raft算法实现的分布式副本管理协议。

Raft的多数派原则，规定了集群的建立无需全员存活，只要存活多数(n个节点，存活节点数>n/2)即可视为集群为正常状态。

这个规定解决了集群中出现的脑裂问题。

比如4台节点组成的集群，其中2台为上海机房，另外2台为北京机房，当上海与北京的网络出现故障，那么一个集群就会分裂为2个集群，这就是脑裂现象。

由于这个规定，导致集群中n台和n+1台节点，他们的容灾能力是一样的（n为奇数），都只能坏一台。使用奇数个节点反而能节省资源。

二、监控

2.1 监控组件比较

常见的Kafka监控如下

  1. CMAK (previously known as Kafka Manager)
    • 优点：监控功能更强大
    • 缺点：a. 重量级 b. 不支持KRaft，参考Kafka 3.3.1 with KRaft Support · Issue #898 · yahoo/CMAK
  2. kafdrop
    • 优点：a. 轻量，开箱即用 b. 支持KRaft
    • 缺点：a. 实时性监控并不好 b. 交互做得并不好，当节点开启身份验证时，会存在严重卡顿情况

2.2 kafdrop使用

首先下载Release 4.0.1 · obsidiandynamics/kafdrop

执行命令，启动监控

java -jar kafdrop-4.0.1.jar --kafka.brokerConnect=10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092

kafdrop本身是一个springboot项目，对于javaer来说是很有好的。其他一些复杂的配置，可以直接下载源码查看application.yml

# 省略以上若干配置。详细内容可自行查看
kafka:brokerConnect: localhost:9092saslMechanism: "PLAIN"securityProtocol: "SASL_PLAINTEXT"truststoreFile: "${KAFKA_TRUSTSTORE_FILE:kafka.truststore.jks}"propertiesFile : "${KAFKA_PROPERTIES_FILE:kafka.properties}"keystoreFile: "${KAFKA_KEYSTORE_FILE:kafka.keystore.jks}"

“${KAFKA_PROPERTIES_FILE:kafka.properties}”

表示获取环境变量KAFKA_PROPERTIES_FILE，如果存在则使用环境变量值，否则使用默认值kafka.properties

比如我开启了授权，那么需要再kafdrop的同级目录下创建kafka.properties，然后再次启动。

security.protocol: SASL_PLAINTEXT
sasl.mechanism: PLAIN
sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";

最终kafdrop监控效果如图。

三、使用

3.1 SpringKafka

3.1.1 使用

创建springboot项目，添加依赖

<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId>
</dependency>

生产者application.yml配置

spring:kafka:
#    bootstrap-servers: 10.0.0.101:9092bootstrap-servers: 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092# 接入授权properties:security.protocol: SASL_PLAINTEXTsasl.mechanism: PLAINsasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";producer:key-serializer: org.apache.kafka.common.serialization.StringSerializervalue-serializer: org.apache.kafka.common.serialization.StringSerializer

消费者application.yml配置

spring:kafka:
#    bootstrap-servers: 10.0.0.10:9092bootstrap-servers: 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092# 接入授权properties:security.protocol: SASL_PLAINTEXTsasl.mechanism: PLAINsasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";consumer:key-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializergroup-id: group # 默认的分组名，消费者都会有所属组

源码参考kafka-cluster-demo/spring-kafka-demo at master · meethigher/kafka-cluster-demo

3.1.2 重试机制

关于SpringKafka的重试机制，未进行深入探索，只参考了文章Spring Kafka：Retry Topic、DLT 的使用与原理 - 知乎。

感觉这篇文章讲得很透彻了。按照该文章的说明，验证了以下两种重试策略

  1. 默认重试策略
  2. 自定义重试策略
默认重试策略

快速重试10次，无间隔时间，如果最后还是失败，则自动commit。

@KafkaListener(topics = "test-retry")
public void test(ConsumerRecord<?, ?> record, Consumer<?, ?> consumer) {int i=1/0;
}

不需要额外配置，这是默认的重试策略。

自定义重试策略

Spring单独提供了RetryableTopic注解，及重试后的回调注解DltHandler。底层逻辑是新建了topic对这些失败的数据进行存储，以及监听这些新建的topic再进行消费，细节的话还是参考文章Spring Kafka：Retry Topic、DLT 的使用与原理 - 知乎。

@KafkaListener(topics = "test-retry")
@org.springframework.kafka.annotation.RetryableTopic(attempts = "5", backoff = @org.springframework.retry.annotation.Backoff(delay = 5000, maxDelay = 10000, multiplier = 2))
public void test(ConsumerRecord<?, ?> record, Consumer<?, ?> consumer) {int i = 1 / 0;
}@DltHandler
public void listenDlt(String in, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,@Header(KafkaHeaders.OFFSET) long offset) {log.error("DLT Received: {} from {} @ {}", in, topic, offset);
}

更多的案例可以参考spring-kafka/samples at main · spring-projects/spring-kafka

3.2 KafkaClient

添加依赖

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>3.3.1</version>
</dependency>

生产者示例

import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.util.Properties;public class Producer {private static Logger log = LoggerFactory.getLogger(Producer.class);public static void main(String[] args) throws Exception {// 配置生产者属性Properties props = new Properties();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092");props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");// 配置 SASL_PLAINTEXT 认证props.put("security.protocol", "SASL_PLAINTEXT");props.put("sasl.mechanism", "PLAIN");props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"alice\" password=\"alice-secret\";");// 创建生产者实例KafkaProducer<String, String> producer = new KafkaProducer<>(props);// 发送消息到指定主题String topic = "meethigher";String key = "timestamp";while (true) {// 发送消息String value= String.valueOf(System.currentTimeMillis());ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);producer.send(record, (metadata, exception) -> log.info("sent key={},value={} to topic={},partition={}", record.key(), record.value(), metadata.topic(), metadata.partition()));System.in.read();}// 关闭生产者
//        producer.close();}
}

消费者示例

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.time.Duration;
import java.util.Collections;
import java.util.Properties;public class Consumer {private static final Logger log = LoggerFactory.getLogger(Consumer.class);public static void main(String[] args) {// 配置消费者属性Properties props = new Properties();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092");props.put(ConsumerConfig.GROUP_ID_CONFIG, "your_consumer_group");props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");// 配置 SASL_PLAINTEXT 认证props.put("security.protocol", "SASL_PLAINTEXT");props.put("sasl.mechanism", "PLAIN");props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"alice\" password=\"alice-secret\";");// 创建消费者实例KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);// 订阅主题String topic = "meethigher";consumer.subscribe(Collections.singletonList(topic));// 消费消息while (true) {ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));records.forEach(record -> {log.info("Consumed record with key {} and value {}", record.key(), record.value());});}}
}

四、参考致谢

Apache Kafka

spring-projects/spring-kafka: Provides Familiar Spring Abstractions for Apache Kafka

Listeners in KAFKA

kafka+Kraft模式集群+安全认证_kafka认证机制_鸢尾の的博客-CSDN博客

我的 Kafka 旅程 - SASL+ACL 认证授权 · 配置 · 创建账号 · 用户授权 · 应用接入 - Sol·wang - 博客园

从k8s集群主节点数量为什么是奇数来聊聊分布式系统 - 知乎

万字长文解析raft算法原理 - 知乎

Spring Kafka：Retry Topic、DLT 的使用与原理 - 知乎

这篇关于KRaft使用SASL_PLAINTEXT进行认证的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/451243。

相关文章

Java学习手册之Filter和Listener使用方法

Java学习手册之Filter和Listener使用方法

《Java学习手册之Filter和Listener使用方法》：本文主要介绍Java学习手册之Filter和Listener使用方法的相关资料,Filter是一种拦截器,可以在请求到达Servl... 目录一、Filter（过滤器）1. Filter 的工作原理2. Filter 的配置与使用二、Listen
阅读更多...
Pandas使用AdaBoost进行分类的实现

Pandas使用AdaBoost进行分类的实现

《Pandas使用AdaBoost进行分类的实现》Pandas和AdaBoost分类算法,可以高效地进行数据预处理和分类任务,本文主要介绍了Pandas使用AdaBoost进行分类的实现,具有一定的参... 目录什么是 AdaBoost？使用 AdaBoost 的步骤安装必要的库步骤一：数据准备步骤二：模型
阅读更多...
使用Pandas进行均值填充的实现

使用Pandas进行均值填充的实现

《使用Pandas进行均值填充的实现》缺失数据（NaN值）是一个常见的问题,我们可以通过多种方法来处理缺失数据,其中一种常用的方法是均值填充,本文主要介绍了使用Pandas进行均值填充的实现,感兴趣的... 目录什么是均值填充？为什么选择均值填充？均值填充的步骤实际代码示例总结在数据分析和处理过程中，缺失数
阅读更多...
如何使用 Python 读取 Excel 数据

如何使用 Python 读取 Excel 数据

《如何使用Python读取Excel数据》：本文主要介绍使用Python读取Excel数据的详细教程,通过pandas和openpyxl,你可以轻松读取Excel文件,并进行各种数据处理操... 目录使用 python 读取 Excel 数据的详细教程1. 安装必要的依赖2. 读取 Excel 文件3. 读
阅读更多...
解决Maven项目idea找不到本地仓库jar包问题以及使用mvn install:install-file

解决Maven项目idea找不到本地仓库jar包问题以及使用mvn install:install-file

《解决Maven项目idea找不到本地仓库jar包问题以及使用mvninstall:install-file》：本文主要介绍解决Maven项目idea找不到本地仓库jar包问题以及使用mvnin... 目录Maven项目idea找不到本地仓库jar包以及使用mvn install:install-file基
阅读更多...
Python使用getopt处理命令行参数示例解析(最佳实践)

Python使用getopt处理命令行参数示例解析(最佳实践)

《Python使用getopt处理命令行参数示例解析(最佳实践)》getopt模块是Python标准库中一个简单但强大的命令行参数处理工具,它特别适合那些需要快速实现基本命令行参数解析的场景,或者需要... 目录为什么需要处理命令行参数？getopt模块基础实际应用示例与其他参数处理方式的比较常见问http
阅读更多...
C 语言中enum枚举的定义和使用小结

C 语言中enum枚举的定义和使用小结

《C语言中enum枚举的定义和使用小结》在C语言里,enum（枚举）是一种用户自定义的数据类型,它能够让你创建一组具名的整数常量,下面我会从定义、使用、特性等方面详细介绍enum,感兴趣的朋友一起看... 目录1、引言2、基本定义3、定义枚举变量4、自定义枚举常量的值5、枚举与switch语句结合使用6、枚
阅读更多...
使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)

使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)

《使用Python从PPT文档中提取图片和图片信息(如坐标、宽度和高度等)》PPT是一种高效的信息展示工具,广泛应用于教育、商务和设计等多个领域,PPT文档中常常包含丰富的图片内容,这些图片不仅提升了... 目录一、引言二、环境与工具三、python 提取PPT背景图片3.1 提取幻灯片背景图片3.2 提取
阅读更多...
使用Python实现图像LBP特征提取的操作方法

使用Python实现图像LBP特征提取的操作方法

《使用Python实现图像LBP特征提取的操作方法》LBP特征叫做局部二值模式,常用于纹理特征提取,并在纹理分类中具有较强的区分能力,本文给大家介绍了如何使用Python实现图像LBP特征提取的操作方... 目录一、LBP特征介绍二、LBP特征描述三、一些改进版本的LBP1.圆形LBP算子2.旋转不变的LB
阅读更多...
Maven的使用和配置国内源的保姆级教程

Maven的使用和配置国内源的保姆级教程

《Maven的使用和配置国内源的保姆级教程》Maven是⼀个项目管理工具,基于POM(ProjectObjectModel,项目对象模型)的概念,Maven可以通过一小段描述信息来管理项目的构建,报告... 目录1. 什么是Maven?2.创建⼀个Maven项目3.Maven 核心功能4.使用Maven H
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2