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。

相关文章

浅析Spring Security认证过程

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程，特意画了如下的类图，包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口，接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti
阅读更多...
中文分词jieba库的使用与实景应用（一）

中文分词jieba库的使用与实景应用（一）

知识星球：https://articles.zsxq.com/id_fxvgc803qmr2.html 目录 一.定义： 精确模式（默认模式）： 全模式： 搜索引擎模式： paddle 模式（基于深度学习的分词模式）： 二 自定义词典 三.文本解析   调整词出现的频率 四. 关键词提取 A. 基于TF-IDF算法的关键词提取 B. 基于TextRank算法的关键词提取
阅读更多...
使用SecondaryNameNode恢复NameNode的数据

使用SecondaryNameNode恢复NameNode的数据

1）需求： NameNode进程挂了并且存储的数据也丢失了，如何恢复NameNode 此种方式恢复的数据可能存在小部分数据的丢失。 2）故障模拟 （1）kill -9 NameNode进程 [lytfly@hadoop102 current]$ kill -9 19886 （2）删除NameNode存储的数据（/opt/module/hadoop-3.1.4/data/tmp/dfs/na
阅读更多...
Hadoop数据压缩使用介绍

Hadoop数据压缩使用介绍

一、压缩原则 （1）运算密集型的Job，少用压缩 （2）IO密集型的Job，多用压缩 二、压缩算法比较 三、压缩位置选择 四、压缩参数配置 1）为了支持多种压缩/解压缩算法，Hadoop引入了编码/解码器 2）要在Hadoop中启用压缩，可以配置如下参数
阅读更多...
Makefile简明使用教程

Makefile简明使用教程

文章目录 规则makefile文件的基本语法：加在命令前的特殊符号：.PHONY伪目标: Makefilev1 直观写法v2 加上中间过程v3 伪目标v4 变量 make 选项-f-n-C Make 是一种流行的构建工具，常用于将源代码转换成可执行文件或者其他形式的输出文件(如库文件、文档等)。Make 可以自动化地执行编译、链接等一系列操作。 规则 makefile文件
阅读更多...
使用opencv优化图片（画面变清晰）

使用opencv优化图片（画面变清晰）

文章目录 需求影响照片清晰度的因素 实现降噪测试代码 锐化空间锐化Unsharp Masking频率域锐化对比测试 对比度增强常用算法对比测试 需求 对图像进行优化，使其看起来更清晰，同时保持尺寸不变，通常涉及到图像处理技术如锐化、降噪、对比度增强等 影响照片清晰度的因素 影响照片清晰度的因素有很多，主要可以从以下几个方面来分析 1. 拍摄设备 相机传感器：相机传
阅读更多...
【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯，先赞后看哦~🎈🎈 🏆 作者简介：景天科技苑 🏆《头衔》：大厂架构师，华为云开发者社区专家博主，阿里云开发者社区专家博主，CSDN全栈领域优质创作者，掘金优秀博主，51CTO博客专家等。 🏆《博客》：Python全栈，前后端开发，小程序开发，人工智能，js逆向，App逆向，网络系统安全，数据分析，Django，fastapi
阅读更多...
pdfmake生成pdf的使用

pdfmake生成pdf的使用

实际项目中有时会有根据填写的表单数据或者其他格式的数据，将数据自动填充到pdf文件中根据固定模板生成pdf文件的需求 文章目录 利用pdfmake生成pdf文件1.下载安装pdfmake第三方包2.封装生成pdf文件的共用配置3.生成pdf文件的文件模板内容4.调用方法生成pdf 利用pdfmake生成pdf文件 1.下载安装pdfmake第三方包 npm i pdfma
阅读更多...
零基础学习Redis(10) -- zset类型命令使用

零基础学习Redis(10) -- zset类型命令使用

zset是有序集合，内部除了存储元素外，还会存储一个score，存储在zset中的元素会按照score的大小升序排列，不同元素的score可以重复，score相同的元素会按照元素的字典序排列。 1. zset常用命令 1.1 zadd  zadd key [NX | XX] [GT | LT]   [CH] [INCR] score member [score member ...]
阅读更多...
业务中14个需要进行A/B测试的时刻[信息图]

业务中14个需要进行A/B测试的时刻[信息图]

在本指南中，我们将全面了解有关 A/B测试 的所有内容。 我们将介绍不同类型的A/B测试，如何有效地规划和启动测试，如何评估测试是否成功，您应该关注哪些指标，多年来我们发现的常见错误等等。 什么是A/B测试？ A/B测试（有时称为“分割测试”）是一种实验类型，其中您创建两种或多种内容变体——如登录页面、电子邮件或广告——并将它们显示给不同的受众群体，以查看哪一种效果最好。 本质上，A/B测
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2