1，创建Kafka主题：
Kafka提供了自带的Kafka-topic脚本用于帮助用户创建主题。
bin/kafka-topic.sh --bootstarp-server broker_host:port --create –topic my_topic --partitions 1 --replication-factor 1
create 表明我们要创建主题，而partitions和replication factor分布设置了主题的分区数以及每个分区下的副本数。

2，查询主题
查询所有主题的列表：/bin/kafka-topic.sh --bootstrap-server broker_host:port --list
查询单个主题的详细数据：/bin/kafka-topic.sh --bootstrap-server broker_host:port --describe --topic <topic name>

3，修改主题
A ：修改分区：/bin/kafka-topic.sh --bootstrap-server broker_host : port --alter --topic <topic_name> --partitions <新分区数>
分区数一定要比原有分区数大。

B ：修改主题级别参数：使用kafka-configs脚本修改对应的参数。
修改主题级别的max.message.bytes ：/bin/kafka-configs.sh --zookeeper zookeeper_host:port --entity-type topic --entity-name <topic_name> --alter --add-config max.message.bytes=10485760

这个命令里使用的 –zookeeper，也可以使用 --bootstrap-server，只是他是用来设置动态参数的。

C ：变更副本数
使用kafka-reassign-partitions 脚本，增加副本数

D ：修改主题限速
这是指设置Leader副本和follower 副本使用的带宽。有时候，需要让某个主题的副本在执行副本同步机制时，不要消耗过多的带宽。
要做到这个需要先设置leader.replication.throttled.rate和follower.replication.throttled.rate

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

E ：主题分区迁移
同样是使用kafka-ressign-partitions脚本。

F ：删除主题
/bin/kafka-topic.sh –bootstrap-server broker_host:port --delete --topic <topic_name>
删除主题的操作是异步的，执行完这条命令不代表主题立即就被删除了，它仅仅是被标记成“已删除”状态而已。Kafka会在后台默默地开启主题删除操作。

4 常见主题错误处理

1：主题删除失败
造成主题删除失败的原因有很多，最常见的原因有两个：副本所在Broker宕机了；待删除主题的部分分区依然在执行迁移过程。

解决：
第一步：手动删除Zookeeper节点/admin/delete_topics 下待删除主题为名的znode。
第二步：手动删除该主题的磁盘上的分区目录。
第三步：在Zookeeper中执行rmr/controller，触发Controller重选举，刷新Controller缓存。
在执行最后一步时，要慎重，因为他可能造成大面积的分区Leader重选举。事实上，仅仅执行前两步也是可以的，只是Controller缓存中没有清空删除主题，不影响使用。

2：_consumer_offset占用太多的磁盘

如果发现这个主题占用了过多的磁盘空间，就要显示的使用jstack 命令查看kafka-log-cleaner-thread前缀线程状态。

减少分区数会带来至少两个明显的问题
1：被删除的分区中的数据怎么处理？
如Fever同学所讲，存在那种只有业务代码才知道将被减少的分区中的数据迁移到哪里的情况，出现这种情况就会丢失数据啦！
2：分区数的变动会触发消费者组重平衡？
重平衡也存在一定的风险，应该尽量避免
此节可作为主题管理操作的索引

老师两个问题想问下，

1、根据__consumer_offsets主题查看消费组提交的位移数据：
bin/kafka-console-consumer.sh --bootstrap-server kafka_host:port --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" --from-beginning
但是执行后一点数据都没有，这个命令什么情况下执行才会有数据出来的。

2、读取主题消息和消费组状态的命令，可以消费其他topic吗，我试了下，只有__consumer_offsets能读取出消息
命令：bin/kafka-console-consumer.sh --bootstrap-server kafka_host:port --topic __consumer_offsets --formatter "kafka.coordinator.group.GroupMetadataManager\$GroupMetadataMessageFormatter" --from-beginning