KRaft使用SASL_PLAINTEXT进行认证

news2025/1/20 10:48:28

需要有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 required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret"
    user_alice="alice-secret";
};

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

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

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

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

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

if [ "x$KAFKA_OPTS" ]; then
    export 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=PLAIN

listeners=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.idcontroller.quorum.votersadvertised.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:9092
  saslMechanism: "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:9092
    bootstrap-servers: 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092
    # 接入授权
    properties:
      security.protocol: SASL_PLAINTEXT
      sasl.mechanism: PLAIN
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

消费者application.yml配置

spring:
  kafka:
#    bootstrap-servers: 10.0.0.10:9092
    bootstrap-servers: 10.0.0.101:9092,10.0.0.102:9092,10.0.0.103:9092
    # 接入授权
    properties:
      security.protocol: SASL_PLAINTEXT
      sasl.mechanism: PLAIN
      sasl.jaas.config: org.apache.kafka.common.security.plain.PlainLoginModule required username="alice" password="alice-secret";
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      group-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 的使用与原理 - 知乎

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.coloradmin.cn/o/1280246.html

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈,一经查实,立即删除!

相关文章

【vue】尚硅谷vue3学习笔记

Vue3快速上手 1.Vue3简介 2020年9月18日&#xff0c;Vue.js发布3.0版本&#xff0c;代号&#xff1a;One Piece&#xff08;海贼王&#xff09;耗时2年多、2600次提交、30个RFC、600次PR、99位贡献者github上的tags地址&#xff1a;https://github.com/vuejs/vue-next/release…

【C语言之 CJson】学CJson看这一篇就够了

文章目录 前言一、下载CJson二、创建一个json2.1 创建json对象cJSON类型详解 2.2 创建键值对2.3 添加嵌套的 JSON 对象2.4 添加数组创建数组添加元素到数组添加数组到obj 2.5 将 JSON 对象转为字符串2.6 释放内存2.7 示例代码 三、解析json3.1 解析json root3.2 把一个key解析出…

SSM框架(六):SpringBoot技术及整合SSM

文章目录 一、概述1.1 简介1.2 起步依赖1.3 入门案例1.4 快速启动 二、基础配置2.1 三种配置文件方式2.2 yaml文件格式2.3 yaml读取数据方式&#xff08;3种&#xff09; 三、多环境开发3.1 yml文件-多环境开发3.2 properties文件-多环境开发3.3 多环境命令行启动参数设置3.4 多…

yolov5 7.0版本部署手机端。通过pnnx导出ncnn。

yolov5 7.0版本部署手机端。通过pnnx导出ncnn。 流程配置ncnn android yolov5导出自己模型的ncnn修改yolo.py文件导出TorchScript文件pnnx转torchscript为ncnn 安卓运行权重路径输入输出anchors 大小类别名generate_proposals方法修改 结果 流程 网络yolov5 的部署已经有很多了…

STM32串口接收不定长数据(接收中断+超时判断)

玩转 STM32 单片机&#xff0c;肯定离不开串口。串口使用一个称为串行通信协议的协议来管理数据传输&#xff0c;该协议在数据传输期间控制数据流&#xff0c;包括数据位数、波特率、校验位和停止位等。由于串口简单易用&#xff0c;在各种产品交互中都有广泛应用。 但在使用串…

基础组件总结(以Element UI组件库为例)

一般对一个组件的使用方式&#xff1a; 每一个组件都有被绑定的数据&#xff0c; &#xff08;1&#xff09;首先会对组件的数据初始化&#xff08;a.data中赋初值 b. 在生命周期函数created或mounted中为变量赋予初值&#xff09; &#xff08;2&#xff09;由于不同组件的…

CSS新手入门笔记整理:CSS边框样式

边框宽度&#xff1a;boder-width 语法 boder-width:像素值&#xff1b; 边框样式&#xff1a;boder-style 语法 boder-style:取值&#xff1b; 属性值 说明 none 无样式 dashed 虚线 solid 实线 边框颜色&#xff1a;boder-color 语法 boder-color:色值&#xf…

MFC 绘制单一颜色圆形、渐变颜色边框圆形、渐变填充圆形以及绘制三角函数正弦函数曲线.

MFC 绘制三种不同圆形以及绘制正弦函数曲线 本文使用visual Studio MFC 平台实现绘制单一颜色圆形、渐变颜色边框圆形、渐变填充圆形以及绘制三角函数正弦函数曲线. 关于基础工程的创建请参考 01-Visual Studio 使用MFC 单文档工程绘制单一颜色直线和绘制渐变颜色的直线 02-vis…

Hisat-Trinity-PASA等组学分析流程

一边学习&#xff0c;一边总结&#xff0c;一边分享&#xff01; 详细教程请访问&#xff1a; 组学分析流程 本期分析流程 Hisat2-SamtoolsTrinity_GG_denovoPASA … 本期教程文章 题目&#xff1a;Genomic insights into local adaptation and future climate-induced vu…

全面的.NET微信网页开发之JS-SDK使用步骤、配置信息和接口请求签名生成详解

JSSDK使用步骤 步骤一:绑定安全域名&#xff1a; 先登录微信公众平台进入“公众号设置”的“功能设置”里填写“JS接口安全域名”。 步骤二:引入JS文件&#xff1a; 在需要调用JS接口的页面引入如下JS文件&#xff0c;&#xff08;支持https&#xff09;&#xff1a;http://…

分治—快速选择算法

文章目录 &#x1f347;215.数组中的第K个最大元素&#x1f348;1. 题目&#x1f349;2. 算法原理&#x1f34a;3. 代码实现 &#x1f34b;LCR 159. 库存管理 III&#x1f34c;1. 题目&#x1f34d;2. 算法原理&#x1f96d;代码实现 &#x1f347;215.数组中的第K个最大元素 …

〖大前端 - 基础入门三大核心之JS篇㊺〗- 定时器和延时器

说明&#xff1a;该文属于 大前端全栈架构白宝书专栏&#xff0c;目前阶段免费&#xff0c;如需要项目实战或者是体系化资源&#xff0c;文末名片加V&#xff01;作者&#xff1a;不渴望力量的哈士奇(哈哥)&#xff0c;十余年工作经验, 从事过全栈研发、产品经理等工作&#xf…

MathType公式编辑器安装教程

一、下载 MathType7是一款可以帮助用户快速完成数学公式编辑的应用软件&#xff0c;这款软件适合在进行教育教学、科研机构、论文写作的时候使用。我们可以直接通过这款软件来获取到大量数学上使用到的函数、数学符号等内容&#xff0c;然后使用这些内容来完成公式编辑。 …

ROS第一个程序——helloworld

目录 一、工作空间的创建 1.创建工作空间并初始化 2.进入 src 创建 ros 包并添加依赖 二、C实现helloworld C源码实现 编辑 ros 包下的 Cmakelist.txt文件 进入工作空间目录并编译 执行 三、python实现helloworld 进入 ros 包添加 scripts 目录并编辑 python 文件 …

【设计模式-4.1】行为型——观察者模式

说明&#xff1a;本文介绍设计模式中行为型设计模式中的&#xff0c;观察者模式&#xff1b; 商家与顾客 观察者模式属于行为型设计模式&#xff0c;关注对象的行为。以商家与顾客为例&#xff0c;商家有商品&#xff0c;顾客来购买商品&#xff0c;如果商家商品卖完了&#…

【Leetcode题单】(01 数组篇)刷题关键点总结02【统计数组中的元素】

【Leetcode题单】&#xff08;01 数组篇&#xff09;刷题关键点总结02【统计数组中的元素】&#xff08;6题&#xff09; 统计数组中的元素645. 错误的集合 Easy697. 数组的度 Easy448. 找到所有数组中消失的数字 Easy442. 数组中重复的数据 Medium41. 缺失的第一个正数 Hard27…

【问题解决】Linux内核编译安装后磁盘空间已满问题

Linux内核编译安装后磁盘空间已满问题解决过程 【注】本文为个人遇到Linux内核经过make&#xff0c;make modules……乃至最后install以后VMware磁盘空间爆炸的情况后&#xff0c;而又不想重装虚拟机&#xff0c;自己找资料实现解决的&#xff0c;文章中很多链接是来自别的博主…

python动态圣诞下雪图

运行图片 代码 import pygame import random# 初始化Pygame pygame.init()# 创建窗口 width, height 800, 600 screen pygame.display.set_mode((width, height)) pygame.display.set_caption(Christmas Tree)# 定义颜色 GREEN (34, 139, 34) RED (255, 0, 0) WHITE (255…

GAMES101:作业2记录

总览 在上次作业中&#xff0c;虽然我们在屏幕上画出一个线框三角形&#xff0c;但这看起来并不是那么的有趣。所以这一次我们继续推进一步——在屏幕上画出一个实心三角形&#xff0c;换言之&#xff0c;栅格化一个三角形。上一次作业中&#xff0c;在视口变化之后&#xff0…

Redis--13--缓存一致性问题

提示&#xff1a;文章写完后&#xff0c;目录可以自动生成&#xff0c;如何生成可参考右边的帮助文档 文章目录 缓存一致性问题1、先更新缓存&#xff0c;再更新DB方案二&#xff1a;先更新DB&#xff0c;再更新缓存方案三&#xff1a;先删缓存&#xff0c;再写数据库推荐1&…