管理多个集群 轻松检查群集状态（主题，消费者，偏移，代理，副本分发，分区分发） 运行首选副本选举 使用选项生成分区分配以选择要使用的代理 运行分区重新分配（基于生成的分配） 使用可选主题配置创建主题（0.8.1.1具有与0.8.2+不同的配置） 删除主题（仅支持0.8.2+并记住在代理配置中设置delete.topic.enable = true） 主题列表现在指示标记为删除的主题（仅支持0.8.2+） 批量生成多个主题的分区分配，并可选择要使用的代理 批量运行重新分配多个主题的分区 将分区添加到现有主题 更新现有主题的配置 可随意开启对broker级别或者对topic级别的JMX轮询 可方便的过滤出没有id 、所有者、延迟或目录等的消费者编译安装包 首先下载编译安装包： wget https://github.com/yahoo/kafka-manager/archive/2.0.0.2.tar.gz #解压
tar -xvf 2.0.0.2.tar.gz #我这边名字为：CMAK-2.0.0.2。咱给他改个名字：kafka-manager-2.0.0.2 mv CMAK-2.0.0.2/ kafka-manager安装sbt包 因为编译的时候需要sbt 如果没有就会报如下的错： 我们来安装下
curl -L https://www.scala-sbt.org/sbt-rpm.repo > sbt-rpm.repo放入yum 放在yum源中然后去安装 sudo mv sbt-rpm.repo /etc/yum.repos.d/ sudo yum install sbt （另外还有一种方法 ） 去获取他的jar放他指定目录 wget http://maven.aliyun.com/nexus/content/repositories/central/org/scala-sbt/sbt-launch/1.1.1/sbt-launch-1.1.1.jar 开始编译kafka-manager 开始编译（必要要求有外网 可以联网）
#进入kafka - manager目录 cd /opt/kafka-manager #开始编译 ./sbt clean dist编译时间比较长 如果有需要直接编译好的包 可以使用如下的。（本文使用的包） 编译好的kafka-manager包 已经编译好的 kafka-manager包： kafka-manager-2.0.0.2 配置kafka-manager配置 解压好 然后修改配置文件 修改conf/application.conf
文件中zk的地址以及启用Kafka-Manager使用账号登录和消费者的配置（由于不能编译红色字体 都用#提醒啦）
kafka-manager.zkhosts="hadoop01: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 #这里改成true
basicAuthentication.enabled=${?KAFKA_MANAGER_AUTH_ENABLED}
basicAuthentication.ldap.enabled=false
basicAuthentication.ldap.enabled=${?KAFKA_MANAGER_LDAP_ENABLED}
basicAuthentication.ldap.server=""
basicAuthentication.ldap.server=${?KAFKA_MANAGER_LDAP_SERVER}
basicAuthentication.ldap.port=389
basicAuthentication.ldap.port=${?KAFKA_MANAGER_LDAP_PORT}
basicAuthentication.ldap.username=""
basicAuthentication.ldap.username=${?KAFKA_MANAGER_LDAP_USERNAME}
basicAuthentication.ldap.password=""
basicAuthentication.ldap.password=${?KAFKA_MANAGER_LDAP_PASSWORD}
basicAuthentication.ldap.search-base-dn=""
basicAuthentication.ldap.search-base-dn=${?KAFKA_MANAGER_LDAP_SEARCH_base_DN}
basicAuthentication.ldap.search-filter="(uid=$capturedLogin$)"
basicAuthentication.ldap.search-filter=${?KAFKA_MANAGER_LDAP_SEARCH_FILTER}
basicAuthentication.ldap.connection-pool-size=10
basicAuthentication.ldap.connection-pool-size=${?KAFKA_MANAGER_LDAP_CONNECTION_POOL_SIZE}
basicAuthentication.ldap.ssl=false
basicAuthentication.ldap.ssl=${?KAFKA_MANAGER_LDAP_SSL}
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/hdp/3.1.5.0-152/kafka/conf/consumer.properties #重新指定这个配置 如下那个注销了
#kafka-manager.consumer.properties.file=${?CONSUMER_PROPERTIES_FILE} #这个配置注销了
添加conf/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创建jass文件 在kafka-manager/conf 的目录创建jaas.conf 然后把keytab文件也cp过来
# Clinet段用于连接zookeeper认证
#KafkaClient用于连接kafka服务器认证
Client{
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/opt/kafka-manager/conf/kafka.service.keytab"
principal="kafka/hadoop01@HADOOP.COM"
serviceName="kafka"
donotprompt=true;
};
KafkaClient{
com.sun.security.auth.module.Krb5LoginModule required
useKeyTab=true
keyTab="/opt/kafka-manager/conf/kafka.service.keytab"
principal="kafka/hadoop01@HADOOP.COM"
serviceName="kafka"
donotprompt=true;
};
启动脚本
添加启动脚本：/opt/kafka-manager/scripts/rm.sh
#！/bin/bash echo '-------------------------------------------------------------------分界线'$(date +%F%t%T)> manager.out # 配置kafka-manager元数据使用的zookeeper,此处必须是用export export ZK_HOSTS=hadoop01:2181 # kafka-manager 路径 MANAGER_HOME=/opt/kafka-manager # 可执行文件路径 KAFKA_MANAGER=$MANAGER_HOME/bin/kafka-manager # 日志位置 APP_HOME=-Dapplication.home=$MANAGER_HOME # 端口 HTTP_PORT=-Dhttp.port=9001 # SASL安全认证 JAAS_CONF=-Djava.security.auth.login.config=$MANAGER_HOME/conf/jaas.conf KRB5_CONF=-Djava.security.krb5.conf=$MANAGER_HOME/conf/krb5.conf nohup $KAFKA_MANAGER $JAAS_CONF $KRB5_CONF $APP_HOME $HTTP_PORT >manager.out 2>&1 & echo "$!" tailf manager.out启动rm.sh 启动的时候调用rm.sh脚本
./rm.sh
17889 这是kafka-manger的pid
访问kafka-manager WEB UI访问http://hadoop01:9001/ 地址
之前配置了账号密码（admin/admin） 如果没有改过是（admin/password）
添加kafk配置当我们第一次打开这个界面的时候 是空白的 要配置 添加集群
配置：
kafka开启JMX这里要注意下，如果开启JMX轮训 ，必须要在kafka启动脚本中打开JMX端口(kafka节点都打开)
vim /usr/hdp/3.1.5.0-152/kafka/bin/kafka-server-start.sh
#打开JMX端口 export JMX_PORT=9999
放开头就行
然后去重启kafka
检测kafka-manager端口lsof -i:9999
因为开启了kerberos 所以选择认证模式。
save保存 然后去查看
查看topic状态界面上可以查看到Broker信息 topic状态
FAQ 编译问题如果编译有问题 ：
获取不到sbt需要的依赖
首先我们把sbt-launch.jar这个包拿下来。默认路径：/root/.sbt/launchers/1.2.8
然后用解压缩文件打开。
我这边用的小茶壶
在sbt下面有个文件：sbt.boot.properties
更改一下repositories ,也就是yum源下载依赖的地方。
[repositories]
local
aliyun: http://maven.aliyun.com/nexus/content/groups/public/
central: http://repo1.maven.org/maven2/
