為了能夠方便的檢視及管理Kafka叢集，yahoo提供了一個基於Web的管理工具（Kafka-Manager）。這個工具可以方便的檢視叢集中Kafka的Topic的狀態（分割槽、副本及訊息量等），支援管理多個叢集、重新分配Partition及建立Topic等功能。

1、編譯（必須要求機器能聯網）

將Github上的原始碼clone下來後進入kafka-manager目錄，執行如下命令編譯

[[email protected] ~]# cd kafka-manager/
[[email protected] kafka-manager]# ./sbt clean dist
 

這裡編譯需要耗費大量的時間，需要耐心等待。

編譯完成後在kafka-manager/target/universal目錄下找到編譯好的壓縮包kafka-manager-1.3.3.21.zip

[[email protected] universal]# pwd
/data/disk1/kafka-manager/target/universal
[[email protected] universal]# ll

到此為止Kafka-Manager編譯完成。

如果機器不能聯網，那麼自己編譯是不可能的了，除非安裝好了sbt環境。

在這裡我使用編譯好的kafka-manager，下載地址：

2、安裝

將上述編譯好的kafka-manager-1.3.3.21.zip壓縮包，拷貝至/opt/cloudera目錄下並解壓

[[email protected] universal]# cp kafka-manager-1.3.3.21.zip /opt/cloudera/
[[email protected] universal]# cd /opt/cloudera/
[[email protected] cloudera]# unzip kafka-manager-1.3.3.21.zip

修改conf/application.conf檔案中zk的地址以及啟用Kafka-Manager使用賬號登入和消費者的配置

application.conf

kafka-manager.zkhosts="beta1:2181"
kafka-manager.zkhosts=${?ZK_HOSTS}
pinned-dispatcher.type="PinnedDispatcher"
pinned-dispatcher.executor="thread-pool-executor"
application.features=["KMClusterManagerFeature","KMTopicManagerFeature","KMPreferredReplicaElectionFeature","KMReassignPartitionsFeature"]

akka {
  loggers = ["akka.event.slf4j.Slf4jLogger"]
  loglevel = "INFO"
}

akka.logger-startup-timeout = 60s

basicAuthentication.enabled=true
basicAuthentication.enabled=${?KAFKA_MANAGER_AUTH_ENABLED}
basicAuthentication.username="admin"
basicAuthentication.username=${?KAFKA_MANAGER_USERNAME}
basicAuthentication.password="admin"
basicAuthentication.password=${?KAFKA_MANAGER_PASSWORD}
basicAuthentication.realm="Kafka-Manager"
basicAuthentication.excluded=["/api/health"] # ping the health of your instance without authentification

kafka-manager.consumer.properties.file=/usr/local/kafka-manager-1.3.3.18/kafka-manager-1.3.3.18/conf/consumer.properties
#kafka-manager.consumer.properties.file=${?CONSUMER_PROPERTIES_FILE}

consumer.properties

security.protocol=SASL_PLAINTEXT
key.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
value.deserializer=org.apache.kafka.common.serialization.ByteArrayDeserializer
sasl.mechanism=GSSAPI
sasl.kerberos.service.name=kafka

3、啟動

命令：

nohup bin/kafka-manager -Dhttp.port=8090 -Djava.security.auth.login.config=/home/xiet/keytab/jaas/jaas.conf > nohup.out 2>&1 &

因為做了kerberos認證，所以需要jaas.conf檔案指定到啟動命令後面。

配置前要保證票據可用！

jaas.conf

KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/home/xiet/keytab/kafka.keytab"
   principal="kafka/[email protected]";
};

Client {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="/home/xiet/keytab/zookeeper.keytab"
  storeKey=true
  useTicketCache=false
  principal="zookeeper/[email protected]";
};

4、介面