Kafka SASL 安全认证

服务端

  • 在服务器节点配置认证文件：
    文件路径：kafka/config/kafka_server_jaas.conf
    文件内容：

    KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin"
user_admin="admin"
user_rex="123456"
user_alice="123456"
user_lucy="123456";
};

    注意配置文件中的两个分号的位置，多一不可，缺一不可。

  • 修改服务器节点的启动配置文件：
    复制 kafka/config/server.properties 为 kafka/config/server-sasl.properties，在文件最末尾添加：

    listeners=SASL_PLAINTEXT://localhost:9092
security.inter.broker.protocol=SASL_PLAINTEXT
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:admin

  • 修改服务器节点的启动脚本：
    复制 kafka/bin/kafka-server-start.sh 为 kafka/bin/kafka-server-start-sasl.sh，将认证信息配置到 kafka 服务器节点的 JVM 启动参数中：

    if [ "x$KAFKA_OPTS" ]; then
    export KAFKA_OPTS="-Djava.security.auth.login.config=/home/weijie/myprojects/kafka_2.12-2.6.0/config/kafka_server_jaas.conf"
fi

  • 启动 zookeeper：./bin/zookeeper-server-start.sh config/zookeeper.properties

  • 以安全认证的方式启动 kafka-server：./bin/kafka-server-start-sasl.sh config/server-sasl.properties

  • 创建一个 topic：./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

生产者 + 消费者

  • 配置 生产者/消费者 的认证信息：
    文件路径：kafka/config/kafka_client_jaas.conf
    文件内容：

    KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
username="admin"
password="admin";
};

    注意配置文件中的两个分号的位置，多一不可，缺一不可。

  • 修改 生产者 的启动配置文件：
    复制 kafka/config/producer.properties 为 kafka/config/producer-sasl.properties
    在文件最末尾添加：

    security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

  • 修改 消费者 的启动配置文件：
    复制 kafka/config/consumer.properties 为 kafka/config/consumer-sasl.properties
    在文件最末尾添加：

    security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

  • 修改生产者的启动脚本：
    复制 kafka/bin/kafka-console-producer.sh 为 kafka/bin/kafka-console-producer-sasl.sh，将认证信息配置到 kafka 生产者的 JVM 启动参数中：

    if [ "x$KAFKA_OPTS" ]; then
  export KAFKA_OPTS="-Djava.security.auth.login.config=/home/weijie/myprojects/kafka_2.12-2.6.0/config/kafka_client_jaas.conf"
fi

  • 修改消费者的启动脚本：
    复制 kafka/bin/kafka-console-consumer.sh 为 kafka/bin/kafka-console-consumer-sasl.sh，将认证信息配置到 kafka 生产者的 JVM 启动参数中：

    if [ "x$KAFKA_OPTS" ]; then
  export KAFKA_OPTS="-Djava.security.auth.login.config=/home/weijie/myprojects/kafka_2.12-2.6.0/config/kafka_client_jaas.conf"
fi

  • 以安全认证的方式启动 kafka-producer：./bin/kafka-console-producer-sasl.sh --broker-list localhost:9092 --topic test --producer.config config/producer-sasl.properties
    发送消息测试：

    生产者发送消息测试.png

  • 以安全认证的方式启动 kafka-consumer：./bin/kafka-console-consumer-sasl.sh --bootstrap-server localhost:9092 --topic test --consumer.config config/consumer-sasl.properties

    消费者接收消息.png

java client

  • maven pom.xml 
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka_2.12</artifactId>
    <version>2.6.0</version>
</dependency>

  • java client 中添加 SASL 设置信息：

    Java client consumer properties配置.png

    注意 sasl.jaas.config 配置中的分号必不可少。

    package kafka;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TestKafkaSasl {

    private static final Logger logger = LoggerFactory.getLogger(TestKafkaSasl.class);
    
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test_group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // sasl.jaas.config的配置, 结尾分号必不可少.
        props.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"admin\" password=\"admin\";");
        props.setProperty("security.protocol", "SASL_PLAINTEXT");
        props.setProperty("sasl.mechanism", "PLAIN");

        @SuppressWarnings("resource")
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        consumer.subscribe(Arrays.asList("test"));
        while (true) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, partition = %d, key = %s, value = %s%n",
                          record.offset(), record.partition(), record.key(), record.value());
                    logger.info("offset = {}, partition = {}, key = {}, value = {}",
                          record.offset(), record.partition(), record.key(), record.value());
                }
            } catch (Exception e) {
                e.printStackTrace();
                logger.error(e.getMessage());
            }
        }
    }
}

    运行测试：

    继续生产数据.png
    java client sasl消费数据测试.png

可能遇到的问题

  • 生产者进程启动报错

    生产者启动报错.png

    解决方案：以安全认证的方式启动 kafka-producer，别忘记添加设置配置文件参数。 ./bin/kafka-console-producer-sasl.sh --broker-list localhost:9092 --topic test --producer.config config/producer-sasl.properties

  • 消费者进程启动报错

    消费者启动报错.png

    解决方案：以安全认证的方式启动 kafka-consumer，别忘记添加设置配置文件参数。 ./bin/kafka-console-consumer-sasl.sh --bootstrap-server localhost:9092 --topic test --consumer.config config/consumer-sasl.properties

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 212,383评论 6赞 493
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 90,522评论 3赞 385
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 157,852评论 0赞 348
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 56,621评论 1赞 284
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 65,741评论 6赞 386
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,929评论 1赞 290
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 39,076评论 3赞 410
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 37,803评论 0赞 268
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 44,265评论 1赞 303
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 36,582评论 2赞 327
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,716评论 1赞 341
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 34,395评论 4赞 333
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 40,039评论 3赞 316
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,798评论 0赞 21
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 32,027评论 1赞 266
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 46,488评论 2赞 361
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 43,612评论 2赞 350