kafka 0.10.0.0 配置SASL_PLAINTEXT_warn sasl configuration failed: javax.security.aut

首先增加kafka-jaas配置文件，修改server.properties，修改启动脚本sh

1、必须配置security.inter.broker.protocol=SASL_PLAINTEXT

原来2.2的时候没有配置这一项，0.11的时候也没有配置，这个版本不配置kafka启动不了，所以增加配置

security.inter.broker.protocol=SASL_PLAINTEXT

listeners=SASL_PLAINTEXT://0.0.0.0:10092
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
security.inter.broker.protocol=SASL_PLAINTEXT
1
2
3
4

2、org.apache.kafka.common.KafkaException: Exception while loading Zookeeper JAAS login context ‘Client’

0.11 中zookper的日志中
WARN SASL configuration failed: javax.security.auth.login.LoginException: No JAAS configuration section named ‘Client’ was found in specified JAAS configuration file: ‘./bin/…/config/kafka_server_jaas.conf’. Will continue connection to Zookeep
er server without SASL authentication, if Zookeeper server allows it. (org.apache.zookeeper.ClientCnxn)
也有这个提示，但是不影响kafka启动，1.10报告的是错误，直接kafka启动不起来

解决方法： 需要配置kafka和zookeeper之间也使用用户名密码

• （1）zookeeper 增加jaas配置文件

Server {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="kafka"
    password="kafka"
    user_kafka="kafka";
};
1
2
3
4
5
6

• （2）zookeeper-server-start.sh

export KAFKA_OPTS="-Djava.security.auth.login.config=file:$base_dir/../config/zk_server_jaas.conf -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationP
rovider -Dzookeeper.requireClientAuthScheme=sasl"
1
2

• （3）kafka_server_jaas.conf

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

1
2
3
4
5
6
7
8
9
10
11
12

请注意：zookeeper的server和kafka的client 都应该用 org.apache.zookeeper.server.auth.DigestLoginModule required

因为：zookeeper does not support SASL Plain, but DigestMD5 is pretty similar.

3、java.io.IOException: Configuration Error: Line : expected [option key]

KafkaServer 配置的 最后没家分号

最后：

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin"
    user_admin="admin";
};
Client {
    org.apache.zookeeper.server.auth.DigestLoginModule required
    username="kafka"
    password="kafka";
};
1
2
3
4
5
6
7
8
9
10
11

4、org.apache.zookeeper.KeeperException$InvalidACLException: KeeperErrorCode = InvalidACL for /brokers/ids

因为 server.properties 中加了如下配置，看网上有人搜因为zookeeper.set.acl=true出的错，所以注释了这部分

#super.users=User:kafka
#authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
#zookeeper.set.acl=true
1
2
3

5、org.I0Itec.zkclient.exception.ZkException: org.apache.zookeeper.KeeperException$InvalidACLException: KeeperErrorCode = InvalidACL

看着和上面的错误比较类似，网上查找资料，找到zookeeper日志异常原因—这不是一个报错故障，只是一个user-level KeeperException。可以忽略不做处理的kafka安装好后，第一次启动。zookeeper日志Error:KeeperErrorCode = NoNode for /config/topics/test，是因为kafka请求访问这个路径，但是这个路径还不存在，zookeeper就抛了这个error。kafka会创建这个topic，然后访问zookeeper里topic对应的路径，zookeeper日志抛出error NodeExists for /config/topics（kafka已经把topic创建好了），路径已经存在了。综上所述，这些error是正常的日志信息，可以忽略。

https://stackoverflow.com/questions/43559328/got-user-level-keeperexception-when-processing

所以忽略了，继续进行，创建了topic成功了，再建立生产者的时候，出现了下面的问题，

6、WARN Error while fetching metadata with correlation id 38 : {1001=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

报错内容：leader不可用
原因分析：原因很多 topic正在被删除 正在进行leader选举
给出的解决方案是：使用kafka-topics脚本检查leader信息进而检查broker的存活情况 尝试重启解决问题
没找到怎么使用kafka-topics脚本检查leader信息，但是看日志，发现了选leader失败的日志。尝试重启
在zookeeper的日志中，发现

7、ERROR Missing AuthenticationProvider for sasl (org.apache.zookeeper.server.PrepRequestProcessor)

在zookeeper-server-start.sh 中设置：
-Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
-Dzookeeper.requireClientAuthScheme=sasl

最终：

export KAFKA_OPTS="-Djava.security.auth.login.config=file:$base_dir/../config/zk_server_jaas.conf -Dzookeeper.authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider -Dzookeeper.requireClientAuthScheme=sasl"

1
2

错误消失，可以成功选择leader

8、consumer-property is not a recognized option

[root@aiot bin]# ./kafka-console-consumer-saal.sh --bootstrap-server 10.221.13.102:10092 --topic 1001 --consumer-property security.protocol=SASL_PLAINTEXT --consumer-property sasl.mechanism=PLAIN

这个命令是在2.2中用的，肯定是当前版本不支持，找到官方对应版本

官方文档：
https://kafka.apache.org/0100/documentation.html#quickstart_consume

Step 5: Start a consumerKafka also has a command line consumer that will dump out messages to standard output.> bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning
This is a message
This is another message

If you have each of the above commands running in a different terminal then you should now be able to type messages into the producer terminal and see them appear in the consumer terminal.All of the command line tools have additional options; running the command with no arguments will display usage information documenting them in more detail.

可以通过执行.sh 不带任何参数，查看这个sh支持的参数

[root@aiot bin]# ./kafka-console-consumer.sh 
The console consumer is a tool that reads data from Kafka and outputs it to standard output.
Option                                  Description                            
------                                  -----------                            
--blacklist <blacklist>                 Blacklist of topics to exclude from    
                                          consumption.                         
--bootstrap-server <server to connect                                          
  to>                                                                          
--consumer.config <config file>         Consumer config properties file.       
--csv-reporter-enabled                  If set, the CSV metrics reporter will  
                                          be enabled                           
--delete-consumer-offsets               If specified, the consumer path in     
                                          zookeeper is deleted when starting up
--enable-systest-events                 Log lifecycle events of the consumer   
                                          in addition to logging consumed      
                                          messages. (This is specific for      
                                          system tests.)                       
--formatter <class>                     The name of a class to use for         
                                          formatting kafka messages for        
                                          display. (default: kafka.tools.      
                                          DefaultMessageFormatter)             
--from-beginning                        If the consumer does not already have  
                                          an established offset to consume     
                                          from, start with the earliest        
                                          message present in the log rather    
                                          than the latest message.             
--key-deserializer <deserializer for                                           
  key>                                                                         
--max-messages <Integer: num_messages>  The maximum number of messages to      
                                          consume before exiting. If not set,  
                                          consumption is continual.            
--metrics-dir <metrics directory>       If csv-reporter-enable is set, and     
                                          this parameter isset, the csv        
                                          metrics will be outputed here        
--new-consumer                          Use the new consumer implementation.   
--property <prop>                       The properties to initialize the       
                                          message formatter.                   
--skip-message-on-error                 If there is an error when processing a 
                                          message, skip it instead of halt.    
--timeout-ms <Integer: timeout_ms>      If specified, exit if no message is    
                                          available for consumption for the    
                                          specified interval.                  
--topic <topic>                         The topic id to consume on.            
--value-deserializer <deserializer for                                         
  values>                                                                      
--whitelist <whitelist>                 Whitelist of topics to include for     
                                          consumption.                         
--zookeeper <urls>                      REQUIRED: The connection string for    
                                          the zookeeper connection in the form 
                                          host:port. Multiple URLS can be      
                                          given to allow fail-over.    
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

果然是不支持，但是有–consumer.config 这个参数，所以修改consumer.properties

[root@at config]# cat consumer.properties 
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
# 
#    http://www.apache.org/licenses/LICENSE-2.0
# 
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# see kafka.consumer.ConsumerConfig for more details

# Zookeeper connection string
# comma separated host:port pairs, each corresponding to a zk
# server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002"
zookeeper.connect=127.0.0.1:10181

# timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000

#consumer group id
group.id=test-consumer-group

#consumer timeout
#consumer.timeout.ms=5000

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

命令变为：
./kafka-console-consumer-saal.sh --zookeeper localhost:10181 --topic 1001 --consumer.config consumer.properties

返回：No brokers found in ZK.

9、No brokers found in ZK.

看到网上有人说：
Can you try the consumer without --zookeeper flag. If you are using the new consumer in 0.10.2 it’s not needed. If you provide --zookeeper then it tries to use the old 0.8 consumer.

翻译：您可以在没有–zookeeper标志的情况下尝试使用消费者吗？ 如果您在0.10.2中使用新使用者，则不需要。 如果提供–zookeeper，则它将尝试使用旧的0.8使用者。

直接去掉–zookeeper肯定是不可以的，因为是REQUIRED ，继续看kafka-console-consumer.sh 参数 有 --new-consumer 这一项
所以修改为：

/kafka-console-consumer-saal.sh --bootstrap-server 127.0.0.1:10092 --topic 1001 --consumer.config …/config/consumer.properties --new-consumer

终于可以成功接收生产者的消息了。

官方文档：https://kafka.apachecn.org/documentation.html