Kafka 如何给集群配置Scram账户认证

前言

很早之前的一篇博客【Kafka+ Centos7服务器集群详细安装教程】 详细的说了下一个真正的集群应该如何搭建Kafka环境，由于当时的需求只是能够使用Kafka服务，因此并没有做别的什么配置。但是随着Kafka不断地使用，仅仅能用已经无法满足业务的安全需求，因此对于Kafka做了一个权限的配置，那么本篇就针对Kafka如何配置账户认账做一个详细的演示，理论知识会稍有涉及但不是重点。更多内容请点击【Apache Kafka API AdminClient 目录】。

官网的权限支持

Kafka本身支持的认证方式有很多，同时支持SSL或者SASL，上面展示的四种是kafka官方推荐的SASL认证方式，如何选择一个合适的认证方式呢？我们可以做一个横向对比。

如果使用SASL/GSSAPI那么需要新搭建Kerberos不太划算；SASL/PLAIN的方式可能会在使用过程中频繁的重启，非常的繁琐；而SASL/OAUTHBEARER属于Kafka新提供的，而且也没有这方面的需求，可以等等市场反应再说。因此综合来说最终选择了SASL/SCRAM的认证方法增强Kafka的安全功能，这也是本篇博客的由来。

Zookeeper配置

Zookeeper集群怎么搭建参考【Zookeeper + Centos7 详细安装教程】，目前笔者用的Zookeeper 已经升级到3.5.8，但是搭建过程是一样的，这里只贴出来Zookeeper的配置文件。首先进入Zookeeper根目录下的/config目录。

给zoo.cfg 添加SASL认证

新添加的两行，用来支持SASL认证。

tickTime=2000 initLimit=10 syncLimit=5 dataDir=/home/data/zookeeper_data dataLogDir=/home/data/zookeeper_log #这里可以不配置，直接删了 clientPort=2181 #新添加的两行在这里 authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider requireClientAuthScheme=sasl server.1=192.168.33.101:2887:3887 server.2=192.168.33.102:2887:3887 server.3=192.168.33.103:2887:3887 server.4=192.168.33.104:2887:3887 server.5=192.168.33.105:2887:3887 

创建Zookeeper认证文件

使用vi zookeeper_jaas.conf命令把下面的内容粘贴进去，保存备用。

Server { 
      org.apache.zookeeper.server.auth.DigestLoginModule required user_super="super1234" user_kafka="kafka1234"; }; 

这里要对这两个参数说明一下：user_super="super1234"这句配置的是超级用户，在Zookeeper里面超级用户默认就是super，后面引号里设置的则是它的密码super1234。因此下面的user_kafka="kafka1234"设置的是Kafka连接Zookeeper要用的账户和密码，这个账户和密码在后面Kafka的配置中还要用，请先记住这点。其意思就是，一个叫做kafka的账户名，密码是kafka1234。前面的user_就是为了识别这个配置是一个账户名用的，吐槽下这个设计真是好奇葩。

启动时加载认证文件

回到/bin目录，使用vi zookeeper-start.sh命令，创建一个新的启动脚本，用于Zookeeper启动时加载认证文件，把下面的内容粘贴进去，保存退出。注意这里的文件路径和启动路径都是在/bin下的，如果你配置的东西不在这里，要修改路径。

export JVMFLAGS="-Djava.security.auth.login.config=../conf/zookeeper_jaas.conf -Dzookeeper.4lw.commands.whitelist=*" ./zkServer.sh start & 

然后使用命令chmod u+x zookeeper-start.sh给脚本文件赋权。

启动

到此如果在/bin目录下直接输入命令sh zookeeper-start.sh就可以启动Zookeeper，有如下字样Zookeeper就是启动成功了。

[root@centos01 bin]# sh zookeeper-start.sh [root@centos01 bin]# ZooKeeper JMX enabled by default Using config: /usr/app-zk-test/apps/apache-zookeeper-3.5.8-bin/bin/../conf/zoo.cfg Starting zookeeper ... STARTED 

如果不放心还可以用ps –ef | grep “zookeeper”来看下启动的现成是不是加载了zookeeper_jaas.conf文件，截取一部分输出，下面红框就是加载的认证文件。也可以用zkCli.sh去连接服务器看看是不是真的可用。那么到此Zookeeper配置启动完毕。

Kafka配置

启动完Zookeeper以后，就要着手进行Kafka相关的配置了。一个可用的Kafka集群怎么搭建的可以参考这篇文章【Kafka+ Centos7服务器集群详细安装教程】，目前笔者的版本已经升级到Kafka 2.7.0，但是这个帖子依然可以使用。首先进入Kafka根目录下的/config目录，笔者先贴上上次博客中所配置的部分。

#服务器id，这个是每个机器的唯一id。每个机器都是独一无二的，习惯性就按顺序来。 broker.id=1 #监听端口，这里也可以用hostname代替，只要host里面配置好了都可以 listeners = PLAINTEXT://192.168.33.101:9092 #存放当前数据的目录，这里所有机器尽量都一样，这样操作起来比较容易 log.dirs=/home/data/kafka-logs #连接Zookeeper。注意这里笔者再Zookeeper里面创建了一个叫做“/kakfa”的节点，所以最后一个ip上跟着 # 一个/kafka，看起来比较整洁。如果没有这样的需求不需要带。 zookeeper.connect=192.168.33.101:2181,192.168.33.102:2181,192.168.33.103:2181,192.168.33.104:2181,192.168.33.105:2181/kafka #如果要直连根节点，请仿照这个 #zookeeper.connect=192.168.33.101:2181,192.168.33.102:2181,192.168.33.103:2181,192.168.33.104:2181,192.168.33.105:2181 

上面就是之前博客中所有的Kafka配置，基本上就是一个裸连能用的配置。首先我们先说下listeners这个参数，这个参数是监听端口，必须要配置的。采用什么样的认证也是通过这个参数后面配置的参数决定的。Kafka默认的参数是PLAINTEXT，也就是我们上面配置的listeners = PLAINTEXT://192.168.33.101:9092这句话。这里的PLAINTEXT就是代表没有任何认证，阿猫阿狗都可以连进来。要说明的是PLAINTEXT和上面介绍的SASL/PLAIN完全不是一个概念，SASL/PLAIN是一种文件认证方式，而PLAINTEXT是一个具体的认证模式。如果要开启认证则需要使用SASL_PLAINTEXT，或者进一步使用SASL_SSL进行双重加密，这个是后话了。

给server.properties添加SASL认证

参数到底应该怎么配置呢？下面就是一个笔者配置的关键参数例子，大家可以按照这个参数去配置自己的Kafka认证。至于其他的参数，大家可以根据实际需求取舍，这里笔者推荐几个重要的参数，实际使用中最好配置上。

# 基础配置如下  broker.id=1 #默认监控端口，设置9092使用SASL_PLAINTEXT协议 listeners=SASL_PLAINTEXT://192.168.33.101:9092 #advertised.listeners控制生产者与消费者接入的端口，如果不设置默认都用listeners，设置9092使用SASL_PLAINTEXT协议 advertised.listeners=SASL_PLAINTEXT://192.168.33.101:9092 log.dirs=/home/data/kafka-logs zookeeper.connect=192.168.33.101:2181,192.168.33.102:2181,192.168.33.103:2181,192.168.33.104:2181,192.168.33.105:2181/kafka # SASL/SCRAM相关配置如下  #Broker内部联络使用的security协议 security.inter.broker.protocol=SASL_PLAINTEXT #Broker内部联络使用的sasl协议，这里也可以配置多个，比如SCRAM-SHA-512,SCRAM-SHA-256并列使用 sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 #Broker允许使用的sasl协议，这里也可以配多个PLAIN,SCRAM-SHA-512,SCRAM-SHA-256 sasl.enabled.mechanisms=PLAIN,SCRAM-SHA-512 #设置zookeeper是否使用ACL zookeeper.set.acl=true #设置ACL类(低于 2.4.0 版本推荐使用 SimpleAclAuthorizer) #authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer #设置ACL类(高于 2.4.0 版本推荐使用 AclAuthorizer) authorizer.class.name=kafka.security.authorizer.AclAuthorizer #设置Kafka超级用户账号，这两个分别对应zookeeper_jaas.conf中的user_super="super1234"和user_kafka="kafka1234"; super.users=User:admin;User:kafka  其他辅助配置，笔者推荐的重要配置  #每条最大消息设置为3MB，超过此size会报错，可以自由调整 replica.fetch.max.bytes= message.max.bytes= #默认的备份数量，可以自由调整 default.replication.factor=2 #默认的partion数量，可以自由调整 num.partitions=3 #是否允许彻底删除topic，低版本这里设置为false则是隐藏topic delete.topic.enable=true #如果topic不存在，是否允许创建一个新的。这里特别推荐设置为false，否则可能会因为手滑多出很多奇奇怪怪的topic出来 auto.create.topics.enable=false 

创建Kafka认证文件

说完服务器的配置，现在还是在/config目录下，我们还需要在启动的时候加载一个认证文件。所以直接vi kafka-broker-jaas.conf创建一个认证文件，粘贴下面内容，保存退出。

KafkaServer { 
         org.apache.kafka.common.security.scram.ScramLoginModule required username="admin" password="admin1234"; }; Client { 
         org.apache.zookeeper.server.auth.DigestLoginModule required username="kafka" password="kafka1234"; }; 

这里要解释一下这两块的内容：首先KafkaServer这里配置的是Kafka服务器本身的超级账户admin和其密码，使用的是ScramLoginModule模式，也就是标题的登陆认证方式。直接使用这个超级账户登陆，整个Kafka集群就相当于对你打开了大门。需要设计一些Kafka工具的时候可以使用，所以好好保存不要泄露了。后面配置的Client是用来登陆Zookeeper使用的，也就是上面我们配置到zookeeper_jaas.conf 文件中的user_kafka="kafka1234"一行所对应的，这里看到登陆Zookeeper要用的账户就是kafka，密码就是kafka1234。这点设计的比较绕，需要多理解理解。

启动时加载认证文件

直接在/config文件下执行vi kafka-starter.sh，这里笔者懒得跳到/bin目录下，直接创建了。粘贴下面内容到文件里，保存退出。然后使用命令chmod u+x kafka-starter.sh给脚本文件赋权。

export KAFKA_OPTS="-Djava.security.auth.login.config=/usr/app-zk-test/apps/kafka_2.13-2.7.0/config/kafka-broker-jaas.conf" export JMX_PORT=9999 export KAFKA_HEAP_OPTS="-Xmx1G -Xms1G" ../bin/kafka-server-start.sh ../config/server.properties & 

第一行KAFKA_OPTS配置的是加载的认证文件的路径；第二行JMX_PORT是监控端口，可以不配置；第三行KAFKA_HEAP_OPTS是配置启动占用内存的，随意调整，也可以不配置用默认的；第四行执行Kafka开始脚本和做好的配置文件。

给Zookeeper中添加超级账户

完成以后跳转到Kafka的/bin目录下，用自带的kafka-configs.sh脚本把Kafka服务器的超级账户添加到Zookeeper中，因为目前(Kafka 2.7.0)来说Kafka账号密码还是存在Zookeeper上的。这一步不需要Kafka启动，但是Zookeeper要启动。

#输入命令创建超级用户 sh kafka-configs.sh --zookeeper 192.168.33.101:2181/kafka --alter --add-config 'SCRAM-SHA-512=[password=admin1234]' --entity-type users --entity-name admin #如果是在Zookeeper根目录下，不需要带“/kafka” sh kafka-configs.sh --zookeeper 192.168.33.101:2181 --alter --add-config 'SCRAM-SHA-512=[password=admin1234]' --entity-type users --entity-name admin 

启动类配置认证文件

当上述步骤都配置完毕以后有些/bin目录下的命令（比如kafka-console-producer.sh）都不能直接使用了，需要带着用户名密码才可以，这就给我们做一些简单的测试造成了很大的麻烦。我们可以通过在启动类中配置认证文件，从而跳过用户名密码的输入，这一步就是让Kafka服务器识别SASL/PLAIN的认证方式。具体做法就是vi kafka-run-class.sh打开这个脚本，然后把下面的一行贴进去，文件开头，文件末尾都可以，不要贴到循环或者if条件语句中就行。保存退出就可以准备启动了，再次提醒认证文件路径要写对。

#粘贴下面这一行到kafka-run-class.sh脚本中 KAFKA_OPTS="-Djava.security.auth.login.config=/usr/app-zk-test/apps/kafka_2.13-2.7.0/config/kafka-broker-jaas.conf" 

启动

重新跳转到/config目录下，执行sh kafka-starter.sh启动Kafka。笔者推荐使用nohup sh kafka-starter.sh后台启动，这样日志就不会打印到前台，还可以做别的事情。那么所有的Kafka机器按照上面的步骤配置一边，然后启动完毕，就可以开始使用了。第一运行会打印下面的started字样，如果不是第一次启动，则会把所有内容加载一遍输出，这个字段就不太好找了。如果是nohup启动，需要去nohup.out中查看。

[2020-08-03 21:14:43,863] INFO Kafka commitId : 21234bee (org.apache.kafka.common.utils.AppInfoParser) [2020-08-03 21:14:43,864] INFO [KafkaServer id=2] started (kafka.server.KafkaServer) 

注意事项

1. 以上步骤凡是涉及到目录和名称都是笔者自己在机器上配置的，不必一定按照笔者的来，只要路径写对了能访问到对应文件，都没有问题。如果按照笔者操作和配置的路径走，一定不会有路径问题。
2. 以上步骤必须全部在Linux(Centos、Ubuntu)系统下操作，切不可图省事儿从外部Windows拖拽或者上传到Linux，可能会因为文件内容的格式不一致导致无法启动Zookeeper或者Kafka。
3. 以上步骤要在所有机器上做一遍，避免某个节点挂了整个服务不可用。

验证程序

public class TestProducer { 
             public static void main(String[] args) { 
             Properties props = new Properties(); props.put("bootstrap.servers", "192.168.33.101:9092"); props.put("acks", "1"); props.put("retries", 3); props.put("batch.size", 16384); props.put("linger.ms", 1); props.put("buffer.memory", ); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "SCRAM-SHA-512"); props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username='easy' password='easy1234';"); /*props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username='admin' password='admin1234';");*/ KafkaProducer<String, String> producer = new KafkaProducer<>(props); for (int i = 0; i < 10; i++) { 
             ProducerRecord<String, String> record = new ProducerRecord<>("my-topic", "topic_" + i, "topic_:" + i); Future<RecordMetadata> metadataFuture = producer.send(record); RecordMetadata recordMetadata = null; try { 
             recordMetadata = metadataFuture.get(); System.out.println("发送成功！"); System.out.println("topic:" + recordMetadata.topic()); System.out.println("partition:" + recordMetadata.partition()); System.out.println("offset:" + recordMetadata.offset()); } catch (Exception e) { 
             System.out.println("发送失败！"); e.printStackTrace(); } } producer.flush(); producer.close(); } } 如果有未授权的用户尝试发送数据会报错 java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512 

Consumer：

public class EasyTestConsumer { 
              public static void main(String[] args) { 
              Properties props = new Properties(); props.put("bootstrap.servers", "192.168.33.101:9092"); props.put("group.id", "aaa"); props.put("enable.auto.commit", "false"); props.put("auto.offset.reset", "earliest"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("security.protocol", "SASL_PLAINTEXT"); props.put("sasl.mechanism", "SCRAM-SHA-512"); props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username='easy' password='easy1234';"); /*props.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username='admin' password='admin1234';");*/ KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); consumer.subscribe(Arrays.asList("my-topic")); while (true) { 
              ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { 
              System.out.printf("partition= %d, offset = %d, key = %s, value = %s\n", record.partition(), record.offset(), record.key(), record.value()); } } } } 如果有未授权的用户尝试读取数据就会报错 Exception in thread "main" org.apache.kafka.common.errors.SaslAuthenticationException: Authentication failed during authentication due to invalid credentials with SASL mechanism SCRAM-SHA-512 如果有未授权的分组尝试读取数据就会报错 Exception in thread "main" org.apache.kafka.common.errors.GroupAuthorizationException: Not authorized to access group: aaa 

认证后赋权

配置好以后除了admin超级账户以外其他所有客户端都无法直接连接因此我们需要对账号进行赋权，可以参照以下命令执行。

创建账号： sh kafka-configs.sh --zookeeper 192.168.33.101:2181/kafka --alter --add-config 'SCRAM-SHA-512=[password=easy1234]' --entity-type users --entity-name easy 添加账号写权限： sh kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.33.101:2181/kafka --add --allow-principal User:easy --operation Read --topic my-topic 添加账号读权限： sh kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.33.101:2181/kafka --add --allow-principal User:easy --operation Read --topic my-topic 创建Group： sh kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=192.168.33.101:2181/kafka --add --allow-principal User: easy --group aaa 

如果要删除，只需要把--add换成--remove即可，或者参考【Apache Kafka API AdminClient 目录】里的内容，使用官方提供的AdminClient写一套Java版本的工具出来，用命令行操作实在太繁琐了。要对Kafka进行更一进步的SSL认证，可以参考【Kafka 如何给集群配置SSL认证】。

可能的报错

这里照例列出笔者在配置过程中遇到的报错。

1. inter.broker.listener.name must be a listener name defined in advertised.listeners.

[2021-02-18 15:50:42,400] ERROR Exiting Kafka due to fatal exception (kafka.Kafka$) java.lang.IllegalArgumentException: requirement failed: inter.broker.listener.name must be a listener name defined in advertised.listeners. The valid options based on currently configured listeners are PLAINTEXT at kafka.server.KafkaConfig.validateValues(KafkaConfig.scala:1781) at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:1756) at kafka.server.KafkaConfig.<init>(KafkaConfig.scala:1312) at kafka.server.KafkaServerStartable$.fromProps(KafkaServerStartable.scala:34) at kafka.Kafka$.main(Kafka.scala:68) at kafka.Kafka.main(Kafka.scala) 

这个报错有2种可能：

1. 没有配置advertised.listeners这个参数，配置上即可；
2. 参数security.inter.broker.protocol=SASL_PLAINTEXT里面配置的内容在listeners 和advertised.listeners里面都没有。因为这两个参数可以给不同的端口配置不同的安全协议，比如listeners=SASL_PLAINTEXT:// 192.168.33.101:9092, PLAINTEXT:// 192.168.33.101:9093, SASL_SSL:// 192.168.33.101:9094。这样用户就可以通过9092端口使用SASL/Scram验证连接，使用9094端口使用SSL+SASL验证连接，以及通过9093端口进行完全不要任何验证的连接，因此PLAINTEXT一定不要在生产环境上配置，太危险了。回到我们的报错，security.inter.broker.protocol配置的内容，必须在listeners里面配置的有才行，比如下面的配置就一定会报这个异常。

listeners= PLAINTEXT:// 192.168.33.101:9093 security.inter.broker.protocol= SASL_PLAINTEXT  下面是一些说明  #listeners和advertised.listeners参数可以根据不同的端口和需求配置不同的认证方式，建议二者保持一致，如果有内外网，就区分端口给不同的上下游 listeners=SASL_PLAINTEXT://192.168.33.101:9092,PLAINTEXT://192.168.33.101:9093,SASL_SSL://192.168.33.101:9094 advertised.listeners=SASL_PLAINTEXT://192.168.33.101:9092,PLAINTEXT://192.168.33.101:9093,SASL_SSL://192.168.33.101:9094 #security.inter.broker.protocol这个参数的值必须配置一个上面有的才行 security.inter.broker.protocol= SASL_PLAINTEXT 

2. ERROR SASL authentication failed using login context ‘Client’ with exception: {}

[2021-02-14 17:47:16,241] ERROR SASL authentication failed using login context 'Client' with exception: { 
             } (org.apache.zookeeper.client.ZooKeeperSaslClient) javax.security.sasl.SaslException: Error in authenticating with a Zookeeper Quorum member: the quorum member's saslToken is null. at org.apache.zookeeper.client.ZooKeeperSaslClient.createSaslToken(ZooKeeperSaslClient.java:312) at org.apache.zookeeper.client.ZooKeeperSaslClient.respondToServer(ZooKeeperSaslClient.java:275) at org.apache.zookeeper.ClientCnxn$SendThread.readResponse(ClientCnxn.java:882) at org.apache.zookeeper.ClientCnxnSocketNIO.doIO(ClientCnxnSocketNIO.java:101) at org.apache.zookeeper.ClientCnxnSocketNIO.doTransport(ClientCnxnSocketNIO.java:363) at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1223) [2021-02-14 17:47:16,243] ERROR [ZooKeeperClient Kafka server] Auth failed. (kafka.zookeeper.ZooKeeperClient) [2021-02-14 17:47:16,244] INFO EventThread shut down for session: 0xb (org.apache.zookeeper.ClientCnxn) [2021-02-14 17:47:16,269] ERROR Fatal error during KafkaServer startup. Prepare to shutdown (kafka.server.KafkaServer) org.apache.zookeeper.KeeperException$AuthFailedException: KeeperErrorCode = AuthFailed for /kafka at org.apache.zookeeper.KeeperException.create(KeeperException.java:130) at org.apache.zookeeper.KeeperException.create(KeeperException.java:54) at kafka.zookeeper.AsyncResponse.maybeThrow(ZooKeeperClient.scala:564) at kafka.zk.KafkaZkClient.createRecursive(KafkaZkClient.scala:1662) at kafka.zk.KafkaZkClient.makeSurePersistentPathExists(KafkaZkClient.scala:1560) at kafka.server.KafkaServer.$anonfun$initZkClient$2(KafkaServer.scala:461) at kafka.server.KafkaServer.$anonfun$initZkClient$2$adapted(KafkaServer.scala:458) at scala.Option.foreach(Option.scala:437) at kafka.server.KafkaServer.initZkClient(KafkaServer.scala:458) at kafka.server.KafkaServer.startup(KafkaServer.scala:233) at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44) at kafka.Kafka$.main(Kafka.scala:82) at kafka.Kafka.main(Kafka.scala) 

这种报错是Kafka无法链接Zookeeper导致的，一般有以下两种原因：

1. 使用windows拖拽的方式会导致有无法看见的结束符，不识别conf文件而造成失败。
2. 或者启动的时候没有用新建的bash文件加载配置文件，而是直接启动服务器脚本，导致没有加载conf文件报错。