Kafka 主题管理

创建主题

创建 Kafka 主题

• create : 创建主题
• partitions : 主题的分区数
• replication-factor : 每个分区下的副本数

bin/kafka-topics.sh \
--bootstrap-server broker_host:port \
--create --topic my_topic_name \
--partitions 1 --replication-factor 1

用 --bootstrap-server 的原因 :

• 用 --zookeeper 会绕过 Kafka 的安全体系。用 --zookeeper ，不受认证体系的约束 , 就能成功创建任意主题
• 用 --bootstrap-server 与集群进行交互，从 Kafka 2.2 后，社区推荐用 --bootstrap-server

查看主题

查询所有主题的列表

bin/kafka-topics.sh \
--bootstrap-server broker_host:port --list

查询单个主题的详细数据

bin/kafka-topics.sh \
--bootstrap-server broker_host:port \
--describe --topic <topic_name>

• --zookeeper : 集群中所有的主题详细数据
• --bootstrap-server : 受到安全认证体系的约束，回权限验证，返回能看到的主题

修改主题

增加分区 :

• --alter : 增加某个主题的分区数
• 指定分区数要 > 原有分区数，否则会抛出 InvalidPartitionsException 异常

bin/kafka-topics.sh \
--bootstrap-server broker_host:port \
--alter --topic <topic_name> \
--partitions < 新分区数 >

设置 max.message.bytes :

• --bootstrap-server : 设置动态参数
• --zookeeper : 设置常规的主题级别参数

bin/kafka-configs.sh \
--zookeeper zookeeper_host:port \
--entity-type topics --entity-name <topic_name> \
--alter --add-config max.message.bytes=10485760

设置副本同步机制的带宽 :

bin/kafka-configs.sh \
--zookeeper zookeeper_host:port \
--alter --add-config 'leader.replication.throttled.rate=104857600,follower.replication.throttled.rate=104857600' \
--entity-type brokers --entity-name 0

为所有副本设置限速 :

bin/kafka-configs.sh \
--zookeeper zookeeper_host:port \
--alter --add-config 'leader.replication.throttled.replicas=*,follower.replication.throttled.replicas=*' \
--entity-type topics --entity-name test

删除主题 :

• 后台异步删除主题

bin/kafka-topics.sh \
--bootstrap-server broker_host:port \ 
--delete  --topic <topic_name>

内部主题

内部主题

• __consumer_offsets : 消费位移
• __transaction_state : 引入事务

修改副本数 :

• 50 个分区对应的副本数

{"version":1, "partitions":[{"topic":"__consumer_offsets","partition":0,"replicas":[0,1,2]}, {"topic":"__consumer_offsets","partition":1,"replicas":[0,2,1]},{"topic":"__consumer_offsets","partition":2,"replicas":[1,0,2]},{"topic":"__consumer_offsets","partition":3,"replicas":[1,2,0]},...{"topic":"__consumer_offsets","partition":49,"replicas":[0,1,2]}
]}

bin/kafka-reassign-partitions.sh \
--zookeeper zookeeper_host:port \
--reassignment-json-file reassign.json --execute

查看消费者组提交的位移数据

bin/kafka-console-consumer.sh \
--bootstrap-server kafka_host:port \
--topic __consumer_offsets \
--formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" \
--from-beginning

查看消费者组的状态

bin/kafka-console-consumer.sh 
--bootstrap-server kafka_host:port 
--topic __consumer_offsets 
--formatter "kafka.coordinator.group.GroupMetadataManager\$GroupMetadataMessageFormatter" 
--from-beginning

异常

__consumer_offsets 占用太多的磁盘，用 jstack 查看 kafka-log-cleaner-thread 的线程状态

• 该线程挂了，无法及时清理此内部主题，重启相应的 Broker
• 注意保留出错日志，可能 Bug 导致，在社区看下

主题删除失败

主题删除失败原因 :

• 副本所在的 Broker 宕机 : 重启 Broker 后，删除会自动恢复
• 待删除主题的部分分区依然在执行迁移过程

主题无法删除的万能方法：

1. 手动删除 ZooKeeper 节点 /admin/delete_topics 下 , 待删除主题的 znode
2. 手动删除该主题在磁盘上的分区目录
3. 在 ZooKeeper 中执行 rmr /controller ，触发 Controller 重选举，刷新 Controller 缓存。 (注意 : 分区 Leader 重选举)