如何为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 required
    user_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端口的通信需要经过认证。

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

  1. sasl.enabled.mechanisms=PLAIN 意思是采用PLAIN这种方式做认证。
  2. listeners中的SASL_PLAINTEXT监听器,表示监听19092端口,并对经过该端口的通信做认证。
  3. 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个步骤服务端就配置好了,下面我们使用客户端验证认证是否生效。

配置客户端

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

  1. kafka命令工具测试
  2. offset explorer图形工具连接kafka
  3. 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");

我们已经为Kafka集群配置好SASL/PLAIN的认证方式。但这个方式的缺点也是显而易见,它只能在启动Kafka前,固定填写好用户名和密码,无法做到灵活修改账号密码。若要修改,只能重启集群。下一篇文章,我们介绍一种可以灵活修改的认证方式,即SASL/SCRAM。


转载请保留本文连接

原文地址:https://blog.csdn.net/weixin_37561180/article/details/136091257

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。

相关推荐


文章浏览阅读4.1k次。kafka认证_kafka认证
文章浏览阅读4.8k次,点赞4次,收藏11次。kafka常用参数_kafka配置
文章浏览阅读1.4k次,点赞25次,收藏10次。Kafka 生产者发送消息的流程涉及多个步骤,从消息的创建到成功存储在 Kafka 集群中。_kafka发送消息流程
文章浏览阅读854次,点赞22次,收藏24次。点对点模型:适用于一对一的消息传递,具有高可靠性。发布/订阅模型:适用于广播消息给多个消费者,实现消息的广播。主题模型:适用于根据消息的主题进行灵活的过滤和匹配,处理复杂的消息路由需求。
文章浏览阅读1.5k次,点赞2次,收藏3次。kafka 自动配置在KafkaAutoConfiguration
文章浏览阅读1.3w次,点赞6次,收藏33次。Offset Explorer(以前称为Kafka Tool)是一个用于管理和使Apache Kafka ®集群的GUI应用程序。它提供了一个直观的UI,允许人们快速查看Kafka集群中的对象以及存储在集群主题中的消息。它包含面向开发人员和管理员的功能。二、环境信息系统环境:windows 10版本:2.2Kafka版本:Kafka2.0.0三、安装和使用3.1 下载Offset Explorer 和安装下载到本地的 .exe文件Next安装路径 ,Next。_offset explorer
文章浏览阅读1.3k次,点赞12次,收藏19次。kafka broker 在启动的时候,会根据你配置的listeners 初始化它的网络组件,用来接收外界的请求,这个listeners你可能没配置过,它默认的配置是listeners=PLAINTEXT://:9092就是告诉kafka使用哪个协议,监听哪个端口,如果我们没有特殊的要求的话,使用它默认的配置就可以了,顶多是修改下端口这块。
文章浏览阅读1.3k次,点赞2次,收藏2次。Kafka 是一个强大的分布式流处理平台,用于实时数据传输和处理。通过本文详细的介绍、使用教程和示例,你可以了解 Kafka 的核心概念、安装、创建 Topic、使用生产者和消费者,从而为构建现代分布式应用打下坚实的基础。无论是构建实时数据流平台、日志收集系统还是事件驱动架构,Kafka 都是一个可靠、高效的解决方案。_博客系统怎么使用kafka
文章浏览阅读3.5k次,点赞42次,收藏56次。对于Java开发者而言,关于 Spring ,我们一般当做黑盒来进行使用,不需要去打开这个黑盒。但随着目前程序员行业的发展,我们有必要打开这个黑盒,去探索其中的奥妙。本期 Spring 源码解析系列文章,将带你领略 Spring 源码的奥秘。本期源码文章吸收了之前 Kafka 源码文章的错误,将不再一行一行的带大家分析源码,我们将一些不重要的分当做黑盒处理,以便我们更快、更有效的阅读源码。废话不多说,发车!
文章浏览阅读1.1k次,点赞14次,收藏16次。一、自动提交offset1、概念Kafka中默认是自动提交offset。消费者在poll到消息后默认情况下,会自动向Broker的_consumer_offsets主题提交当前主题-分区消费的偏移量2、自动提交offset和手动提交offset流程图3、在Java中实现配置4、自动提交offset问题自动提交会丢消息。因为如果消费者还没有消费完poll下来的消息就自动提交了偏移量,那么此时消费者挂了,于是下一个消费者会从已经提交的offset的下一个位置开始消费消息。_kafka中自动提交offsets
文章浏览阅读1.6k次。如果生产者发送消息的速度超过发送到服务器的速度,则会导致生产者空间不足,这个时候KafkaProducer的send()方法调用要么被阻塞,要么抛出异常,这个取决于参数max.block.ms的配置,此参数的默认值为60000,即60秒。在默认情况下,生产者发送的消息是未经压缩的。如果应用程序调用send()方法的速度超过生产者将消息发送给服务器的速度,那么生产者的缓冲空间可能会被耗尽,后续的send()方法调用会等待内存空间被释放,如果在max.block.ms之后还没有可用空间,就抛出异常。_kafka producer 参数
文章浏览阅读2.9k次,点赞3次,收藏10次。kafka解决通信问题_kafka3.6
文章浏览阅读1.5k次,点赞9次,收藏11次。上面都配置完了之后可以先验证下,保证数据最终到ck,如果有问题,需要再每个节点调试,比如先调试nginx->rsyslog ,可以先不配置kafka 输出,配置为console或者文件输出都可以,具体这里就不写了。这里做了一个类型转换,因为nginx,request-time 单位是s,我想最终呈现在grafana 中是ms,所以这里做了转换,当然grafana中也可以做。kafka 相关部署这里不做赘述,只要创建一个topic 就可以。
文章浏览阅读1.4k次,点赞22次,收藏16次。Kafka中的enable-auto-commit和auto-commit-interval配置_auto-commit-interval
文章浏览阅读742次。thingsboard规则链调用外部 kafka_thingsboard kafka
文章浏览阅读1.3k次,点赞18次,收藏22次。Kafka_简介
文章浏览阅读1.1k次,点赞16次,收藏14次。在数据库系统中有个概念叫事务,事务的作用是为了保证数据的一致性,意思是要么数据成功,要么数据失败,不存在数据操作了一半的情况,这就是数据的一致性。在很多系统或者组件中,很多场景都需要保证数据的一致性,有的是高度的一致性。特别是在交易系统等这样场景。有些组件的数据不一定需要高度保证数据的一致性,比如日志系统。本节从从kafka如何保证数据一致性看通常数据一致性设计。
文章浏览阅读1.4k次。概述介绍架构发展架构原理类型系统介绍类型hive_table类型介绍DataSet类型定义Asset类型定义Referenceable类型定义Process类型定义Entities(实体)Attributes(属性)安装安装环境准备安装Solr-7.7.3安装Atlas2.1.0Atlas配置Atlas集成HbaseAtlas集成SolrAtlas集成KafkaAtlas Server配置Kerberos相关配置Atlas集成HiveAtlas启动Atlas使用Hive元数据初次导入Hive元数据增量同步。_atlas元数据管理
文章浏览阅读659次。Zookeeper是一个开源的分布式服务管理框架。存储业务服务节点元数据及状态信息,并负责通知再 ZooKeeper 上注册的服务几点状态给客户端。
文章浏览阅读1.4k次。Kafka-Kraft 模式架构部署_kafka kraft部署