集群Kafka配置SASL用户名密码认证

转载请注明出处：http://blog.csdn.net/u012842205/article/details/73188684

本文中配置的kafka 集群为三节点，Zookeeper有4节点。两个集群相互独立。

Apache Kafka v2.11-0.10.0.0

Apache Zookeeper v3.4.8

Kafka配置SASL PLAIN用于完成基本的用户名密码身份认证。

一、Zookeeper集群配置SASL

zookeeper所有节点都是对等的，只是各个节点角色可能不相同。以下步骤所有的节点配置相同。

1、zoo.cfg文件配置

添加如下配置：

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

2、编写JAAS文件

这个文件定义需要链接到Zookeeper服务器的用户名和密码。JAAS配置节默认为Server：

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required 
        username="admin" 
        password="admin-secret" 
        user_kafka="kafka-sec" 
        user_producer="prod-sec";
};

配置文件我命名为zk_server_jaas.conf，并放在部署目录的conf/下。文件中定义了身份认证类（org.apache.kafka.common.security.plain.PlainLoginModule），可以看到这个认证类是kafka命名空间，也就是需要加入kafka的插件，所以下面一步非常重要。
这个文件中定义了两个用户，一个是kafka，一个是producer（user_<username>可以定义多个用户，等于的值就是用户密码），这些用user_<username>配置出来的用户都可以提供给生产者程序和消费者程序认证使用。还有两个属性，username和password，其中username是配置Zookeeper节点之间内部认证的用户名，password是对应的密码。

3、向zookeeper添加Kafka认证插件

由于Zookeeper的认证机制是使用插件，这个插件只要支持JAAS即可。Kafka需要链接到Zookeeper，直接使用Kafka的认证插件。这个插件类也包含在kafka-clients中（Maven项目）。将依赖的几个jar加入Zookeeper启动的classpath即可。如下是kafka-clients-0.10.0.1相关jar，包括其依赖：

kafka-clients-0.10.0.1.jar
lz4-1.3.0.jar
slf4j-api-1.7.21.jar
slf4j-log4j12-1.7.21.jar
snappy-java-1.1.2.6.jar

我的做法比较直接，在Zookeeper部署根目录下创建一个路径for_sasl，将上述所有jar文件复制到这个路径下，再修改bin/zkEnv.sh配置文件，这个文件主要负责加载一些启动Zookeeper有关的环境变量，输入参数。

for i in "$ZOOBINDIR"/../for_sasl/*.jar; do
  CLASSPATH="$i:$CLASSPATH"
done
SERVER_JVMFLAGS=" -Djava.security.auth.login.config=$ZOOCFGDIR/zk_server_jaas.conf "

逻辑比较简单，先将for_sasl目录下的所有jar文件追加到CLASSPATH变量，再设置一个JVM参数给SERVER_JVMFLAGS变量，这两个变量都会在Zookeeper启动时传给JVM。具体可以查看脚本源码。

4、配置其他节点

照1到3步骤配置剩余的zookeeper节点。

5、启动所有节点

将所有zookeeper节点的Quorum进程开启：bin/zkServer.sh start，查看zookeeper日志，看看之后所有节点是否都能稳定运行，再试试bin/zkCli.sh链接所有节点。是否都能通。

二、Kafka集群配置SASL

确保以上zookeeper配置成功后，开始配置Kafka。Kafka所有节点也是对等的，所以下面步骤的配置在所有节点上都相同。

1、创建JAAS配置文件

定义链接Kafka Broker时所需要的用户名密码及broker各个节点之间相互通信的用户名密码，这部分配置定义在KafkaServer节，文件如下：

KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-sec"
    user_admin="admin-sec"
    user_producer="prod-sec"
    user_consumer="cons-sec";
};

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

另外Client配置节主要配置了broker到Zookeeper的链接用户名密码。

先解释KafkaServer，使用user_<name>来定义多个用户，供客户端程序（生产者、消费者程序）认证使用，可以定义多个，后续配置可能还可以根据不同的用户定义ACL，这部分内容超出本文范围。这是目前我对配置的理解。上例我定义了三个用户，一个是admin，一个是producer，一个是consumer，等号后面是对应用户的密码（如user_producer定义了用户名为producer，密码为prod-sec的用户）。再选择一个用户，用于Kafka内部的各个broker之间通信，这里我选择admin用户，对应的密码是admin-sec。

Client配置节则容易理解得多，主要是broker链接到zookeeper，从上文的Zookeeper JAAS文件中选择一个用户，填写用户名和密码即可。

2、配置server.properties

listeners=SASL_PLAINTEXT://vubuntuez1:9092
security.inter.broker.protocol=SASL_PLAINTEXT  
sasl.enabled.mechanisms=PLAIN  
sasl.mechanism.inter.broker.protocol=PLAIN  
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true

这里注意listeners配置项，将主机名部分（本例主机名是vubuntuez1）替换成当前节点的主机名。其他在各个节点的配置一致。注意，allow.everyone.if.no.acl.found这个配置项默认是false，若不配置成true，后续生产者、消费者无法正常使用Kafka。要么配置成true，这样必须对每个认证用户都配置ACL，比较麻烦，我直接设置true，当没有找到ACL配置时，允许所有的访问操作。

3、配置KAFKA_OPTS环境变量

与伪分布式节点配置KAFKA_OPTS变量相同，主要是为了传递JVM变量java.security.auth.login.config，指定JAAS配置文件。然后kafka-run-class.sh会负责将这个KAFKA_OPTS变量传递给JVM。所以启动kafka broker的命令可以与这个设置变量操作写在同一个脚本下：

#!/bin/bash
ROOT=`dirname $0`
export KAFKA_OPTS=" -Djava.security.auth.login.config=$ROOT/config/kafka_server_jaas.conf "
$ROOT/bin/kafka-server-start.sh -daemon $ROOT/config/server.properties

4、配置其他节点

配置剩余的kafka broker节点，注意server.properties的listeners配置项，上文有提到。

5、启动所有的kafka节点

启动完成后查看日志，看看是否能稳定运行，不怕抛出异常。运行正常后继续下面的操作。注意，一旦配置Kafka认证后，kafka自带的console-producer和console-consumer都无法使用了，具体安全配置请查看我的另一篇博客： Kafka安全认证SASL下附带工具的配置使用。

三、简单的Producer编写

以下所有内容除一部分参数值（比如密码）不同外，其他与伪分布式节点配置SASL的生产者、消费者代码修改流程一致。0.9以后的kafka版本生产者API没有问题，可以直接使用，0.9前的生产者API貌似不支持安全，这个问题具体查看官方文档，更具体的我不了解。

0.10后的kafka，生产者代码我不写了，关键说说需要增加的配置操作。

1、添加配置
JAVA程序的KafkaProducer初始化需要传入Properties，向Properties实例（假设是props）中加入如下配置项：

props.setProperty ("security.protocol", "SASL_PLAINTEXT");
props.setProperty ("sasl.mechanism", "PLAIN");

2、编写JAAS配置文件

这里我们用kafka_client_jaas.conf

KafkaClient {
    org.apache.kafka.common.security.plain.PlainLoginModule required
            username="producer"
            password="prod-sec";
};

配置节KafkaClient，默认的是这个，这个名称也可以修改，自行查官方文档。我们使用alice用户。高版本的kafka，其生产者消费者只需要直接链接kafka broker即可，所以不用配置zookeeper。

3、指定JAAS配置文件

同理，通过变量java.security.auth.login.config传递给JVM。Kafka的client组件会调用 System.getProperty()这个JDK函数获取到JAAS配置文件路径，所以我们直接使用System.setProperty()设置这个变量为上面创建的kafka_client_jaas.conf文件位置即可。

4、启动生产者

提议最好设置一个log4j配置文件，并且设置日志级别为DEBUG，这样就能看见是否正常写入数据。

四、简单的Consumer编写

步骤同producer。