管理多个集群
轻松检查群集状态（主题，消费者，偏移，代理，副本分发，分区分发）
运行首选副本选举
使用选项生成分区分配以选择要使用的代理
运行分区重新分配（基于生成的分配）
使用可选主题配置创建主题（0.8.1.1具有与0.8.2+不同的配置）
删除主题（仅支持0.8.2+并记住在代理配置中设置delete.topic.enable = true）
主题列表现在指示标记为删除的主题（仅支持0.8.2+）
批量生成多个主题的分区分配，并可选择要使用的代理
批量运行重新分配多个主题的分区
将分区添加到现有主题
更新现有主题的配置
可随意开启对broker级别或者对topic级别的JMX轮询
可方便的过滤出没有id 、所有者、延迟或目录等的消费者

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

curl -L https://www.scala-sbt.org/sbt-rpm.repo > sbt-rpm.repo

#进入kafka - manager目录
cd /opt/kafka-manager
#开始编译
./sbt clean dist

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} #这个配置注销了

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

# 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;
};

#！/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