登录
当前位置： 首页 > 工具软件 > Kraft > 使用案例 >

Kafka基于Kraft下的权限控制

雷曜灿
2023-12-01

Kafka基于Kraft下的权限控制

本文基于kafka的版本 3.2.0, 之前的版本无法使用本文所提到的方法。

本文方法对kafka源代码有修改
修改部分如下(metadata\src\main\java\org\apache\kafka\metadata\authorizer\StandardAuthorizerData.java):

    void addAcl(Uuid id, StandardAcl acl) {
        try {
            StandardAcl prevAcl = aclsById.putIfAbsent(id, acl);
            if (prevAcl != null) {
                log.warn("An ACL with ID " + id + " already exists.");
//                throw new RuntimeException("An ACL with ID " + id + " already exists.");
            }
            else if (!aclsByResource.add(acl)) {
                aclsById.remove(id);
                log.warn("Unable to add the ACL with ID " + id +" from aclsByResource");
                // throw new RuntimeException("Unable to add the ACL with ID " + id +
                //     " to aclsByResource");
            }
            else if (log.isTraceEnabled()) {
                log.trace("Added ACL " + id + ": " + acl);
            }
        } catch (Throwable e) {
            log.error("addAcl error", e);
 //           throw e;
        }
    }

    void removeAcl(Uuid id) {
        try {
            StandardAcl acl = aclsById.remove(id);
            if (acl == null) {
                log.warn("ID " + id + " not found in aclsById.");
//                throw new RuntimeException("ID " + id + " not found in aclsById.");
            }
            else if (!aclsByResource.remove(acl)) {
                log.warn("Unable to remove the ACL with ID " + id +" from aclsByResource");
               // throw new RuntimeException("Unable to remove the ACL with ID " + id +
                //    " from aclsByResource");
            }
            else if (log.isTraceEnabled()) {
                log.trace("Removed ACL " + id + ": " + acl);
            }
        } catch (Throwable e) {
            log.error("removeAcl error", e);
            //throw e;
        }
    }

实现作用是把抛出异常换为了输出警告，抛出异常的方式会导致kafka启动的时候无法正常启动，至于为什么kafka启动的时候要执行添加/删除 acl 的操作，暂时还不清楚。无法正常启动时出现的异常如下：

Jul 28 15:29:06 kafka-server-start.sh[123334]: [2022-07-28 15:29:06,133] ERROR [StandardAuthorizer 1] addAcl error (org.apache.kafka.metadata.authorizer.Stand
Jul 28 15:29:06 kafka-server-start.sh[123334]: java.lang.RuntimeException: An ACL with ID eK5n22NLQOeOHTT3gcnf7w already exists.
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizerData.addAcl(StandardAuthorizerData.java:169)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizer.addAcl(StandardAuthorizer.java:83)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$19(BrokerMetadataPublisher.scala:234)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.util.LinkedHashMap$LinkedEntrySet.forEach(LinkedHashMap.java:671)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18(BrokerMetadataPublisher.scala:232)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18$adapted(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.publish(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener.kafka$server$metadata$BrokerMetadataListener$$publish(BrokerMet
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2(BrokerMetadataListener.scala:
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2$adapted(BrokerMetadataListene
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.run(BrokerMetadataListener.scala:119)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:121)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:200)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:173)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.lang.Thread.run(Thread.java:748)
Jul 28 15:29:06 kafka-server-start.sh[123334]: [2022-07-28 15:29:06,139] ERROR [BrokerMetadataPublisher id=1] Error publishing broker metadata at OffsetAndEpo
Jul 28 15:29:06 kafka-server-start.sh[123334]: java.lang.RuntimeException: An ACL with ID eK5n22NLQOeOHTT3gcnf7w already exists.
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizerData.addAcl(StandardAuthorizerData.java:169)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizer.addAcl(StandardAuthorizer.java:83)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$19(BrokerMetadataPublisher.scala:234)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.util.LinkedHashMap$LinkedEntrySet.forEach(LinkedHashMap.java:671)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18(BrokerMetadataPublisher.scala:232)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18$adapted(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.publish(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener.kafka$server$metadata$BrokerMetadataListener$$publish(BrokerMet
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2(BrokerMetadataListener.scala:
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2$adapted(BrokerMetadataListene
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.run(BrokerMetadataListener.scala:119)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:121)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:200)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:173)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.lang.Thread.run(Thread.java:748)
Jul 28 15:29:06 kafka-server-start.sh[123334]: [2022-07-28 15:29:06,143] ERROR [BrokerMetadataListener id=1] Unexpected error handling HandleCommitsEvent (kaf
Jul 28 15:29:06 kafka-server-start.sh[123334]: java.lang.RuntimeException: An ACL with ID eK5n22NLQOeOHTT3gcnf7w already exists.
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizerData.addAcl(StandardAuthorizerData.java:169)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.metadata.authorizer.StandardAuthorizer.addAcl(StandardAuthorizer.java:83)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$19(BrokerMetadataPublisher.scala:234)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.util.LinkedHashMap$LinkedEntrySet.forEach(LinkedHashMap.java:671)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18(BrokerMetadataPublisher.scala:232)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.$anonfun$publish$18$adapted(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataPublisher.publish(BrokerMetadataPublisher.scala:221)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener.kafka$server$metadata$BrokerMetadataListener$$publish(BrokerMet
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2(BrokerMetadataListener.scala:
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.$anonfun$run$2$adapted(BrokerMetadataListene
Jul 28 15:29:06 kafka-server-start.sh[123334]: at scala.Option.foreach(Option.scala:437)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at kafka.server.metadata.BrokerMetadataListener$HandleCommitsEvent.run(BrokerMetadataListener.scala:119)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventContext.run(KafkaEventQueue.java:121)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.handleEvents(KafkaEventQueue.java:200)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at org.apache.kafka.queue.KafkaEventQueue$EventHandler.run(KafkaEventQueue.java:173)
Jul 28 15:29:06 kafka-server-start.sh[123334]: at java.lang.Thread.run(Thread.java:748)

安装

  1. 从官网上下载3.2.0的安装包 ,并解压
    下载地址: https://www.apache.org/dyn/closer.cgi?path=/kafka/3.2.0/kafka_2.13-3.2.0.tgz
tar -xzf kafka_2.13-3.2.0.tgz
cd kafka_2.13-3.2.0
  1. 替换kafka-metadata-3.2.0.jar

基于上面提到的修改代码，重新构建后生成kafka-metadata-3.2.0.jar，替换掉libs/kafka-metadata-3.2.0.jar

# 备份官方的 kafka-metadata-3.2.0.jar
# 一定要把这个包从libs中拿出来
mv libs/kafka-metadata-3.2.0.jar ./
# 然后把自己build的jar包放进去
mv /root/kafka-3.2.0-src/metadata/build/libs/kafka-metadata-3.2.0.jar/kafka-metadata-3.2.0.jar libs/kafka-metadata-3.2.0.jar
  1. 修改配置文件

config/kraft/server.properties:

process.roles=broker,controller
node.id=1
# 修改这里，ip替换为实际ip
controller.quorum.voters=1@<ip1>:9093,2@<ip2>:9093,3@<ip4>:9093
# listeners 的PLAINTEXT要修改为SASL_PLAINTEXT
listeners=SASL_PLAINTEXT://<ip1>:9092,CONTROLLER://<ip1>:9093
# 这里也是PLAINTEXT要修改为SASL_PLAINTEXT
inter.broker.listener.name=SASL_PLAINTEXT
# 这里也是PLAINTEXT要修改为SASL_PLAINTEXT
advertised.listeners=SASL_PLAINTEXT://<ip1>:9092
controller.listener.names=CONTROLLER
# 这里 CONTROLLER:PLAINTEXT修改为 CONTROLLER:SASL_PLAINTEXT
listener.security.protocol.map=CONTROLLER:SASL_PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
# 这里，修改为要存放log的地方(实际存放的应该是kafka的数据，log在kafka安装目录的log文件夹下)
log.dirs=/data/kafka_3.2.0/log
num.partitions=1
num.recovery.threads.per.data.dir=2
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
# 认证方式，用了最简单的PLAIN，缺点是不能动态添加用户
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
sasl.mechanism=PLAIN
# 禁用了自动创建topic
auto.create.topics.enable = false
# 设置必须授权才能用
allow.everyone.if.no.acl.found=false
# 设置超级管理员
super.users=User:admin
# 这个是3.2.0版本新引入的认证方式，可以参考 https://cwiki.apache.org/confluence/display/KAFKA/KIP-801%3A+Implement+an+Authorizer+that+stores+metadata+in+__cluster_metadata
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
# 集群间认证时用的认证方式
sasl.mechanism.controller.protocol=PLAIN

config/kraft/jaas.conf

KafkaServer {
   org.apache.kafka.common.security.plain.PlainLoginModule required
   username="admin"
   password="password"
   user_admin="password"
   user_test="test";
};
  • username/password 表示了认证时用的用户。
  • suer_admin=“password”,这个表示一个用户名为admin用户，密码是password，这个必须要有一个，且要这一个跟上面的username和password保持一致。
  • user_test=“test” 是第二个用户，表示的是用户名为test的账户，密码为test。

service(/usr/lib/systemd/system/kafka.service)

默认kafka的启动方式是通过命令行管理，这里做了一个service用于控制kafka的启动与停止，也作为守护进程。

[Unit]
Description=kafka server daemon

[Service]
Type=simple
# 这里是指定了 jaas.conf文件，用于启用用户认证
Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/data/kafka_3.2.0/package/kafka_2.13-3.2.0/config/kraft/jaas.conf"
# 启动命令
ExecStart=/data/kafka_3.2.0/package/kafka_2.13-3.2.0/bin/kafka-server-start.sh /data/kafka_3.2.0/package/kafka_2.13-3.2.0/config/kraft/server.properties
ExecReload=/bin/kill -HUP $MAINPID
# 停止命令
ExecStop=/data/kafka_3.2.0/package/kafka_2.13-3.2.0/bin/kafka-server-stop.sh
KillMode=process
Restart=on-failure
RestartSec=42s

[Install]
WantedBy=multi-user.target
  1. 生成集群clusterid
./bin/kafka-storage.sh random-uuid
./bin/kafka-storage.sh format -t <uuid> -c ./config/kraft/server.properties
  1. 启动kafka
systemctl daemon-reload
systemctl start kafka

命令行中使用

  1. 先创建一个用于client的认证文件

vim sasl.properties

# 配置上一个用户
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required  username="admin"  password="password";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAI

执行命令式，后面都要带上 --command-config ./sasl.properties来进行用户认证
2. 创建两个topic

# 创建 topic create-for-test 
bin/kafka-topics.sh --bootstrap-server localhost:9092  --create  --topic create-for-test --partitions 1 --replication-factor 1  --command-config ./sasl.properties
# 创建 topic admin-create-test
bin/kafka-topics.sh --bootstrap-server localhost:9092  --create  --topic admin-create-test --partitions 1 --replication-factor 1  --command-config ./sasl.properties
# 查看topic
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list --command-config ./sasl.properties
  1. 为topic create-for-test ，用test赋读权限
bin/kafka-acls.sh  --bootstrap-server localhost:9092 --add --allow-principal User:test --operation Read --topic create-for-test --command-config ./sasl.properties
  1. 切换到test用户，查看topic
# 修改用户，把admin改成test
vim sasl.properties
# 查看所有topic，应该只能看到 create-for-test
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list --command-config ./sasl.properties

java中使用

package org.example;



import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.util.Properties;
import java.util.Collections;
import java.util.UUID;

/**
 * Hello world!
 *
 */
public class App 
{
    public static void main( String[] args )
    {
        String username = "test";
        String password = "test";
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "<ip1>:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, UUID.randomUUID().toString());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 这里配置认证协议
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        // 认证方式
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        // 认证用户
        String saslJaasConfig = String.format("org.apache.kafka.common.security.plain.PlainLoginModule required \nusername=\"%s\" \npassword=\"%s\";", username, password);
        props.put(SaslConfigs.SASL_JAAS_CONFIG, saslJaasConfig);

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        System.out.printf(consumer.listTopics().toString());
        consumer.close();
    }
}
 类似资料：

相关阅读

asp.net core项目mvc权限控制：分配权限vue-router 控制路由权限的实现理解Java访问权限控制Zookeepr 如何进行权限控制？Spring MVC整合Shiro权限控制的方法

相关文章

GitLab CI权限MySQL user权限表GitLab用户权限获得访问权限MySQL其它权限表

相关问答

基于Spring Security URL的授权节点中的用户角色和基于权限的访问。js如何控制JSF中的访问和权限？运行kafka服务器的权限被拒绝如何在Spring Boot中实现基于角色权限的系统

相关文档

Keras 基于 Python 的深度学习库Duang 基于配置自动生成 CMSKeras 基于 Python 的深度学习库中文文档Knative入门 构建基于 Kubernetes 的现代化 Serverless 应用深度学习入门：基于 Python 的理论与实现
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.