繼 搭建免認證kafka單機 之後由於業務需要，搭建了基於SASL認證的kafka偽叢集環境。本次同樣使用的是 kafka_2.10-0.10.1.0.tgz 版本的kafka，整合zookeeper，只需要對此進行配置即可，無需單獨安裝。

一、準備工作

1、環境：ubuntu16
2、安裝包： 開啟此連結 選擇下載的路徑和相應的資料包，我使用的是 kafka_2.11-2.1.0.tgz 版本
3、安裝jdk
jdk分為以下幾種：jre、openjdk、 oracle jdk，這裡我們要安裝的是oracle jdk（推薦安裝），過程：

add-apt-repository ppa:webupd8team/java
apt-get update
apt-get install oracle-java8-installer
apt-get install oracle-java8-set-default
 

二、安裝kafka

1、 解壓安裝包

直接解壓之後，如目錄結構：

2、修改基礎配置項

kafka和zookeeper在需要認證的情況下，需要對其認證方式和基礎配置內容進行配置，主要是修改和新增認證配置。
需要修改的配置項均在config/目錄下，包括以下個：

• server.properties
• zookeeper.properties

我們依次說明其作用和配置項的代表的含義及如何修改配置。
1）server.properties
該配置項是kafka伺服器的配置資訊，目的是配置kafka的監聽服務和配置認證方式。開啟該檔案，新增以下欄位：

listeners=SASL_PLAINTEXT://172.20.36.103:9092
# 使用的認證協議
security.inter.broker.protocol=SASL_PLAINTEXT
# SASL機制
sasl.enabled.mechanisms=PLAIN
sasl.mechanism.inter.broker.protocol=PLAIN
# 完成身份驗證的類
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
# 如果沒有找到ACL（訪問控制列表）配置，則允許任何操作。
allow.everyone.if.no.acl.found=true
 

如下圖所示。

2）zookeeper.properties
該配置項是zookeeper的配置資訊，新增SASL的認證方式，開啟該檔案，在檔案末尾直接新增：

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

如下圖：

到此，kafka和zookeeper的配置項已經修改完成。

3、新增認證配置項

上面2、段落完成了對kafka和zookeeper的認證配置之後，我們還需要配置認證的詳細資訊。原因諸如zookeeper只允許完成認證的kafka-broker訪問，kafka的broker之間也通過認證互相訪問，kafka的生產者和消費者客戶端同樣必須是認證過的才能訪問kafka。

需要配置的項有兩個（名稱可以自定義），分別是：

• zookeeper對kafka叢集的認證配置：kafka_zoo_jaas.conf
• kafka叢集對其他連線broker的認證和kafka儲存zookeeper認證密碼的配置：kafka_server_jaas.conf

依次說明：
1） kafka_zoo_jaas.conf
該JAAS配置檔案提供身份認證外掛，並對登入zookeeper的kafka認證使用者名稱和密碼，建立kafka_zoo_jaas.conf檔案，內容如下：

ZKServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="admin-secret"
    user_admin="admin-secret";
};

這裡配置的應用名稱為ZKServer，zookeeper預設的應用名稱為Server，如何修改後面具體說明。
欄位解釋：
a. ZooKeeper是一個分散式的，開放原始碼的分散式應用程式協調服務。多個zookeeper服務互相通訊業需要身份的認證，由於我們只使用了單個的zookeeper服務，因此我們只定義了一個使用者，若有多個需求則也定義多個。下面就是zookeeper內部認證的使用者名稱和密碼。
username=“admin”
password=“admin-secret”
b. zookeeper對登入的Kafka叢集身份認證定義稍有不同，使用user_使用者名稱=密碼的方式。
user_admin=“admin-secret”
定義了一個使用者名稱為admin，密碼為admin-secret的使用者。
2）kafka_server_jaas.conf
該JAAS配置檔案提供兩個功能：

• kafka的各broker之間的認證；
• 和zookeeper伺服器的認證（用於zookeeper伺服器登入的使用者名稱和密碼）

建立kafka_server_jaas.conf檔案，內容如下：

KafkaServer {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="geting"
    password="geting"
    user_geting="geting"
    user_alice="geting";
};

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

a. Client 用於登入zookeeper時的認證，和本節1）中的使用者名稱密碼一致。
b. KafkaServer 兩個功能，使用者kafka內部broker之間的認證和定義用於kafka客戶端的認證。
username=“geting”
password=“geting”
上面定義，用於kafka各broker之間的認證。

user_geting=“geting”
user_alice=“geting”;
上面定義了兩個kafka客戶端的身份認證，使用者名稱為geting密碼為geting的客戶端和使用者名稱為alice密碼為geting的客戶端。

到此為止，我們已經完成了對zookeeper和kafka的身份認證配置。

4.配置啟動時需要的準備環境變數

通過上面的幾個步驟之後，我們完成了基礎的一些配置，要能啟動成功還需要對環境變數進行配置。這裡的環境變數其實是傳遞給java虛擬機器的引數，使用者JVM完成身份的認證，不是實際意義上的環境變數，為了描述方便我們以下段落均稱為環境變數。
1）zookeeper環境變數設定和啟動
zookeeper環境變數我們添加了兩個，一個是啟動時候的的身份認證JAAS檔案的路徑，另一個是我們修改了Zookeeper的預設應用名稱（3-1中應用名被修改為ZKServer），所以變數為：

-Djava.security.auth.login.config=$ROOT/config/kafka_zoo_jaas.conf -Dzookeeper.sasl.serverconfig=ZKServer

Djava.security.auth.login.config 為身份認證JAAS配置檔案
Dzookeeper.sasl.serverconfig 為修改應用的名稱

整合環境變數和啟動zookeeper時使用的zookeeper-server-start.sh這個指令碼，我們重新寫一個指令碼，用於啟動zookeeper並且新增我們定義的環境變數。
啟動zookeeper指令碼 start_zookeeper.sh，後臺啟動並且輸出日誌到logs下的zookeeper.log日誌檔案中：

#!/bin/bash
ROOT=`dirname $0`
export KAFKA_OPTS=" -Djava.security.auth.login.config=$ROOT/config/kafka_zoo_jaas.conf -Dzookeeper.sasl.serverconfig=ZKServer "
$ROOT/bin/zookeeper-server-start.sh $ROOT/config/zookeeper.properties >> $ROOT/logs/zookeeper.log 2>&1 &

執行該指令碼啟動zookeeper服務。

2）kafka環境變數設定和啟動
和上面所述一致，需要傳遞給JVM的引數為kafka各broker之間身份認證、kafka客戶端身份認證的配置檔案。

 -Djava.security.auth.login.config=$ROOT/config/kafka_server_jaas.conf

啟動kafka服務指令碼 start_kafka.sh，後臺啟動且輸出日誌到logs目錄下的kafka.log日誌檔案中：

#!/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 $ROOT/config/server.properties >> $ROOT/logs/kafka.log 2>&1 &

執行該指令碼啟動kafka服務。
到此，我們完成了kafka服務的所有配置工作，不出意外的話已經成功啟動基於sasl認證的kafka偽叢集服務。

5.測試

該部分的配置檔案只是針對在linux環境下使用kafka自帶的shell指令碼測試時使用，若使用第三方的庫諸如librdkafka或者python的kafka庫測試時是不需要該部分的。

使用kafka自帶的shell指令碼測試資料生產和消費是否正常。需要注意的是由於我們是需要SASL認證的，所以我們在使用指令碼測試前需要做一些準備工作，準備三個配置項（為了方便我直接將身份認證配置檔案寫到了config/目錄下，需要說明的是，不需要必須在config目錄下）：

• 新建消費者/生產者登入kafka時的身份認證配置檔案: kafka_client_jaas.conf

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

使用者名稱和密碼需要與3-2)中配置保持一致,檔名稱可以自定義。

• 修改生產者生產時配置檔案
  在config/目錄下的producer.properties檔案末尾新增認證方式

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

• 修改消費者消費時配置檔案
  在config/目錄下的consumer.properties檔案末尾新增認證方式

security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN

和4段落一樣，為了方便我們這裡也設定環境變數，呼叫bin/kafka-console-producer.sh和bin/kafka-console-consumer.sh寫成shell指令碼，方便使用
1）消費資料指令碼 start_consumer.sh

#!/bin/bash
ROOT=`dirname $0`
export KAFKA_OPTS=" -Djava.security.auth.login.config=$ROOT/config/kafka_client_jaas.conf"
$ROOT/bin/kafka-console-consumer.sh --bootstrap-server 172.20.36.103:9092 --topic test --from-beginning --consumer.config $ROOT/config/consumer.properties

執行指令碼的結果：

2）生產資料指令碼 start_producer.sh

#!/bin/bash
ROOT=`dirname $0`
export KAFKA_OPTS=" -Djava.security.auth.login.config=$ROOT/config/kafka_client_jaas.conf"
$ROOT/bin/kafka-console-producer.sh --broker-list 172.20.36.103:9092 --topic test --producer.config $ROOT/config/producer.properties

執行指令碼的結果：

到此，完成了所有的搭建工作。

參考資料：
https://blog.csdn.net/u012842205/article/details/73188534
http://www.orchome.com/553
https://metron.apache.org/current-book/metron-sensors/bro-plugin-kafka/index.html