登录
当前位置： 首页 > 知识库问答 >
问题：

KafkaSASL_PLAINSCRAM在Spring启动消费者失败

酆乐湛
2023-03-14

我尝试了kafka身份验证SASL_PLAINTEXT/SCRAM，但在Spring Boot中身份验证失败。

我尝试更改SASL_PLAINTEXT/PLAIN，它正在工作。但是SCRAM是身份验证失败SHA-512和SHA-256

做了很多不同的事情，但它不起作用……我该如何修复它？

代理日志

broker1       | [2020-12-31 02:57:37,831] INFO [SocketServer brokerId=1] Failed authentication with /172.29.0.1 (Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512) (org.apache.kafka.common.network.Selector)
broker2       | [2020-12-31 02:57:37,891] INFO [SocketServer brokerId=2] Failed authentication with /172.29.0.1 (Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512) (org.apache.kafka.common.network.Selector)

Spring启动日志

2020-12-31 11:57:37.438  INFO 82416 --- [  restartedMain] o.a.k.c.s.authenticator.AbstractLogin    : Successfully logged in.
2020-12-31 11:57:37.497  INFO 82416 --- [  restartedMain] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 2.6.0
2020-12-31 11:57:37.499  INFO 82416 --- [  restartedMain] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 62abe01bee039651
2020-12-31 11:57:37.499  INFO 82416 --- [  restartedMain] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1609383457495
2020-12-31 11:57:37.502  INFO 82416 --- [  restartedMain] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-Test-Consumer-1, groupId=Test-Consumer] Subscribed to topic(s): test
2020-12-31 11:57:37.508  INFO 82416 --- [  restartedMain] o.s.s.c.ThreadPoolTaskScheduler          : Initializing ExecutorService
2020-12-31 11:57:37.528  INFO 82416 --- [  restartedMain] o.s.b.w.embedded.tomcat.TomcatWebServer  : Tomcat started on port(s): 8080 (http) with context path ''
2020-12-31 11:57:37.546  INFO 82416 --- [  restartedMain] i.m.k.p.KafkaProducerScramApplication    : Started KafkaProducerScramApplication in 2.325 seconds (JVM running for 3.263)
2020-12-31 11:57:37.833  INFO 82416 --- [ntainer#0-0-C-1] o.apache.kafka.common.network.Selector   : [Consumer clientId=consumer-Test-Consumer-1, groupId=Test-Consumer] Failed authentication with localhost/127.0.0.1 (Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512)
2020-12-31 11:57:37.836 ERROR 82416 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-Test-Consumer-1, groupId=Test-Consumer] Connection to node -1 (localhost/127.0.0.1:9091) failed authentication due to: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512
2020-12-31 11:57:37.837  WARN 82416 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-Test-Consumer-1, groupId=Test-Consumer] Bootstrap broker localhost:9091 (id: -1 rack: null) disconnected
2020-12-31 11:57:37.842 ERROR 82416 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Consumer exception

java.lang.IllegalStateException: This error handler cannot process 'org.apache.kafka.common.errors.SaslAuthenticationException's; no record information is available
    at org.springframework.kafka.listener.SeekUtils.seekOrRecover(SeekUtils.java:151) ~[spring-kafka-2.6.4.jar:2.6.4]
    at org.springframework.kafka.listener.SeekToCurrentErrorHandler.handle(SeekToCurrentErrorHandler.java:113) ~[spring-kafka-2.6.4.jar:2.6.4]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.handleConsumerException(KafkaMessageListenerContainer.java:1425) ~[spring-kafka-2.6.4.jar:2.6.4]
    at org.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer.run(KafkaMessageListenerContainer.java:1122) ~[spring-kafka-2.6.4.jar:2.6.4]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]
Caused by: org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512

我的docker-compose.yml

...
...

  zookeeper3:
    image: confluentinc/cp-zookeeper:6.0.1
    hostname: zookeeper3
    container_name: zookeeper3
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper3:2183
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://zookeeper:2183
      ZOOKEEPER_CLIENT_PORT: 2183
      ZOOKEEPER_TICK_TIME: 2000
      ZOOKEEPER_SERVER_ID: 3
      KAFKA_OPTS: "-Djava.security.auth.login.config=/etc/kafka/secrets/sasl/zookeeper_jaas.conf \ 
          -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider \
          -Dzookeeper.authProvider.2=org.apache.zookeeper.server.auth.DigestAuthenticationProvider \
          -Dquorum.auth.enableSasl=true \
          -Dquorum.auth.learnerRequireSasl=true \
          -Dquorum.auth.serverRequireSasl=true \
          -Dquorum.auth.learner.saslLoginContext=QuorumLearner \
          -Dquorum.auth.server.saslLoginContext=QuorumServer \
          -Dquorum.cnxn.threads.size=20 \
          -DrequireClientAuthScheme=sasl"
    volumes: 
      - /etc/kafka/secrets/sasl:/etc/kafka/secrets/sasl

  broker1:
    image: confluentinc/cp-kafka:6.0.1
    hostname: broker1
    container_name: broker1
    depends_on:
      - zookeeper1
      - zookeeper2
      - zookeeper3
    ports:
      - "9091:9091"
      - "9101:9101"
      - "29091:29091"
    expose: 
      - "29090"
    environment:
      KAFKA_OPTS: "-Dzookeeper.sasl.client=true -Djava.security.auth.login.config=/etc/kafka/secrets/sasl/kafka_server_jaas.conf"
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper1:2181,zookeeper2:2182,zookeeper3:2183'
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT,SASL_PLAINHOST:SASL_PLAINTEXT
      KAFKA_LISTENERS: INSIDE://:29090,OUTSIDE://:29091,SASL_PLAINHOST://:9091
      KAFKA_ADVERTISED_LISTENERS: INSIDE://broker1:29090,OUTSIDE://localhost:29091,SASL_PLAINHOST://localhost:9091
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_SECURITY_INTER_BROKER_PROTOCAL: SASL_PLAINTEXT
      KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-512
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAINTEXT
    volumes: 
      - /etc/kafka/secrets/sasl:/etc/kafka/secrets/sasl
  broker2:
    image: confluentinc/cp-kafka:6.0.1
    hostname: broker2
    container_name: broker2
    depends_on:
      - zookeeper1
      - zookeeper2
      - zookeeper3
    ports:
      - "9092:9092"
      - "9102:9102"
      - "29092:29092"
    expose: 
      - "29090"
    environment:
      KAFKA_OPTS: "-Dzookeeper.sasl.client=true -Djava.security.auth.login.config=/etc/kafka/secrets/sasl/kafka_server_jaas.conf"
      KAFKA_BROKER_ID: 2
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper1:2181,zookeeper2:2182,zookeeper3:2183'
      KAFKA_INTER_BROKER_LISTENER_NAME: INSIDE
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT,SASL_PLAINHOST:SASL_PLAINTEXT
      KAFKA_LISTENERS: INSIDE://:29090,OUTSIDE://:29092,SASL_PLAINHOST://:9092
      KAFKA_ADVERTISED_LISTENERS: INSIDE://broker2:29090,OUTSIDE://localhost:29092,SASL_PLAINHOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9102
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_SECURITY_INTER_BROKER_PROTOCAL: SASL_PLAINTEXT
      KAFKA_SASL_ENABLED_MECHANISMS: SCRAM-SHA-512
      KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAINTEXT
    volumes: 
      - /etc/kafka/secrets/sasl:/etc/kafka/secrets/sasl

kaka服务器jaas.conf

KafkaServer {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="admin"
    password="password"
    user_admin="password"
    user_client="password";
};

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

KafkaClient {
    org.apache.kafka.common.security.scram.ScramLoginModule required
    username="client"
    password="password";
};

zookeeper_jaas.conf

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    user_admin="password";
};
QuorumServer {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       user_admin="password";
};
 
QuorumLearner {
       org.apache.zookeeper.server.auth.DigestLoginModule required
       username="admin"
       password="password";
};

消费者配置.java

private static final String BOOTSTRAP_ADDRESS = "localhost:9091,localhost:9092";
private static final String JAAS_TEMPLATE = "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"%s\" password=\"%s\";";

public Map<String, Object> consumerConfigs() {
    Map<String, Object> props = new HashMap<>();
    
    String jaasCfg = String.format(JAAS_TEMPLATE, "client", "password");
    
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_ADDRESS);
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "1000");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "Test-Consumer");
    props.put("sasl.jaas.config", jaasCfg);
    props.put("sasl.mechanism", "SCRAM-SHA-512");
    props.put("security.protocol", "SASL_PLAINTEXT");
    return props;
}

@Bean
public ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory =
      new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}

共有1个答案

方河
2023-03-14

解决了。

因为我没有在zookeeper里添加用户信息。添加此代码。

  zookeeper-add-kafka-users:
    image: confluentinc/cp-kafka:6.0.1
    container_name: "zookeeper-add-kafka-users"
    depends_on:
      - zookeeper1
      - zookeeper2
      - zookeeper3
    command: "bash -c 'echo Waiting for Zookeeper to be ready... && \
                          cub zk-ready zookeeper1:2181 120 && \
                          cub zk-ready zookeeper2:2182 120 && \
                          cub zk-ready zookeeper3:2183 120 && \
                          kafka-configs --zookeeper zookeeper1:2181 --alter --add-config 'SCRAM-SHA-512=[iterations=4096,password=password]' --entity-type users --entity-name admin && \
                          kafka-configs --zookeeper zookeeper1:2181 --alter --add-config 'SCRAM-SHA-512=[iterations=4096,password=password]' --entity-type users --entity-name client '"
    environment:
      KAFKA_BROKER_ID: ignored
      KAFKA_ZOOKEEPER_CONNECT: ignored
      KAFKA_OPTS: -Djava.security.auth.login.config=/etc/kafka/secrets/sasl/kafka_server_jaas.conf
    volumes:
      - /home/mysend/dev/docker/kafka/sasl:/etc/kafka/secrets/sasl

KafkaSASL_PLAIN SCRAM
如果不使用泊坞窗可以使用命令

bin/kafka-configs --zookeeper localhost:2181 --alter --add-config 'SCRAM-SHA-256=[password=admin-secret],SCRAM-SHA-512=[password=admin-secret]' --entity-type users --entity-name admin
 类似资料：
  • kafka生产者/消费者重新启动后，消费者未收到消息

    我们有一个制作人 在开发过程中，我重新部署了producer应用程序，并做了一些更改。但在此之后，我的消费者没有收到任何消息。我尝试重新启动消费者，但没有成功。问题可能是什么和/或如何解决？ 消费者配置： 生产者配置： 编辑2： 5分钟后，消费者应用程序死亡，但以下情况除外：

  • 在Spring kafka消费者启动后运行一次方法

    我有一个使用Spring kafka库的Spring启动应用程序的消费者。我想为每个消费者线程设置租户上下文，即在创建每个线程时调用一个方法（创建时每个线程只调用一次）。目前，我已将其添加到listner中，其中对方法有@KafkaListner注释，但它每次轮询并处理每条记录时都会调用它。我想在消费者线程启动时调用此方法一次。如果我们有任何这样的事情，请在这里帮助我。

  • Spring Boot Kafka：由于NoSuchBeanDefinitionException无法启动消费者

    Spring Boot版本:1.5.2.发行版 下面是异常跟踪：

  • Kafka消费者在重新启动时跳过消息

    我有一个Kafka集群正在运行，当重新启动应用程序（消费者）时，它会跳过一些在应用程序关闭时推送到主题的消息。 当应用程序启动时，我可以看到它读取带有偏移量的消息，然后将偏移量推送到。然后当应用程序关闭时，带有偏移量的消息被推送到主题。重启应用程序后，它读取并将其偏移量设置为，因此跳过。 这是我的配置：

  • 消费者DefaultMQPushConsumer启动主要流程

    主要内容：1 创建DefaultMQPushConsumer实例,2 subscribe订阅,3 start启动消费者,3.1 copySubscription拷贝订阅关系,4 小结基于RocketMQ release-4.9.3，深入的介绍了消费者DefaultMQPushConsumer启动主要流程源码。 此前我们学习了Producer和Broker的启动源码，以及Producer发送消息源码和Broker接收存储消息的源码，现在，我们来学习Consumer的启动以及消费消息的源码。Cons

  • 在启动Spring kafka-消费者之前运行一个方法

    谁能建议在spring启动其kafka消费者之前如何运行初始化我的应用程序的方法？我正在使用spring的@KafkaListener注释创建一个kafka消费者

相关问答

在启动Spring kafka-消费者之前运行一个方法Spring Cloud Data Flow Kinises示例消费者失败Kafka消费者Spring靴Spring启动Kafka不起作用-消费者没有收到消息Kafka消费者：消费者与消费者的抵消滞后

相关文章

消费者驱动的契约（CDC）是什么？Kafka 消费者是否可以消费指定分区消息？简述消费者与消费组之间的关系什么是kafka消费者组？开始：一个生产者众多消费者

相关阅读

Producer生产者启动讯飞消费者java日常实习53min消费者负载均衡服务RebalanceService入口华为消费者bg 交互设计二面ConsumeMessageOrderlyService顺序消费消息

相关工具

费尔进销存费尔模板引擎费尔PHP模板引擎猫国建设者知启蒙内核

相关文档

jMonkeyEngine 初学者教程GitBook 开发者手册PHP 开发者实践EOS 开发者指南YodaOS 开发者文档
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.