Kafka基于Kraft下的權(quán)限控制
本文基于kafka的版本 3.2.0, 之前的版本無法使用本文所提到的方法抡驼。
本文方法對(duì)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;
}
}
實(shí)現(xiàn)作用是把拋出異常換為了輸出警告，拋出異常的方式會(huì)導(dǎo)致kafka啟動(dòng)的時(shí)候無法正常啟動(dòng)爆存，至于為什么kafka啟動(dòng)的時(shí)候要執(zhí)行添加/刪除 acl 的操作，暫時(shí)還不清楚。無法正常啟動(dòng)時(shí)出現(xiàn)的異常如下：
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)
安裝
- 從官網(wǎng)上下載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
- 替換
kafka-metadata-3.2.0.jar
基于上面提到的修改代碼，重新構(gòu)建后生成kafka-metadata-3.2.0.jar，替換掉libs/kafka-metadata-3.2.0.jar
# 備份官方的 kafka-metadata-3.2.0.jar
# 一定要把這個(gè)包從libs中拿出來
mv libs/kafka-metadata-3.2.0.jar ./
# 然后把自己build的jar包放進(jìn)去
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
- 修改配置文件
config/kraft/server.properties:
process.roles=broker,controller
node.id=1
# 修改這里共螺，ip替換為實(shí)際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的地方(實(shí)際存放的應(yīng)該是kafka的數(shù)據(jù)情竹，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
# 認(rèn)證方式璃谨，用了最簡(jiǎn)單的PLAIN，缺點(diǎn)是不能動(dòng)態(tài)添加用戶
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN
sasl.mechanism=PLAIN
# 禁用了自動(dòng)創(chuàng)建topic
auto.create.topics.enable = false
# 設(shè)置必須授權(quán)才能用
allow.everyone.if.no.acl.found=false
# 設(shè)置超級(jí)管理員
super.users=User:admin
# 這個(gè)是3.2.0版本新引入的認(rèn)證方式鲤妥，可以參考 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
# 集群間認(rèn)證時(shí)用的認(rèn)證方式
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 表示了認(rèn)證時(shí)用的用戶。
- suer_admin="password",這個(gè)表示一個(gè)用戶名為admin用戶拱雏，密碼是password棉安，這個(gè)必須要有一個(gè)，且要這一個(gè)跟上面的username和password保持一致铸抑。
- user_test="test" 是第二個(gè)用戶贡耽，表示的是用戶名為test的賬戶，密碼為test。
service(/usr/lib/systemd/system/kafka.service)
默認(rèn)kafka的啟動(dòng)方式是通過命令行管理蒲赂，這里做了一個(gè)service用于控制kafka的啟動(dòng)與停止阱冶，也作為守護(hù)進(jìn)程。
[Unit]
Description=kafka server daemon
[Service]
Type=simple
# 這里是指定了 jaas.conf文件滥嘴，用于啟用用戶認(rèn)證
Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/data/kafka_3.2.0/package/kafka_2.13-3.2.0/config/kraft/jaas.conf"
# 啟動(dòng)命令
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
- 生成集群clusterid
./bin/kafka-storage.sh random-uuid
./bin/kafka-storage.sh format -t <uuid> -c ./config/kraft/server.properties
- 啟動(dòng)kafka
systemctl daemon-reload
systemctl start kafka
命令行中使用
- 先創(chuàng)建一個(gè)用于client的認(rèn)證文件
vim sasl.properties
# 配置上一個(gè)用戶
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="password";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAI
執(zhí)行命令式木蹬，后面都要帶上 --command-config ./sasl.properties來進(jìn)行用戶認(rèn)證
- 創(chuàng)建兩個(gè)topic
# 創(chuàng)建 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
# 創(chuàng)建 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
- 為topic create-for-test ，用test賦讀權(quán)限
bin/kafka-acls.sh --bootstrap-server localhost:9092 --add --allow-principal User:test --operation Read --topic create-for-test --command-config ./sasl.properties
- 切換到test用戶若皱，查看topic
# 修改用戶镊叁，把a(bǔ)dmin改成test
vim sasl.properties
# 查看所有topic，應(yīng)該只能看到 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");
// 這里配置認(rèn)證協(xié)議
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
// 認(rèn)證方式
props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
// 認(rèn)證用戶
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();
}
}
