一、kafka-topics.sh操作

1、查看topic列表

在kafka安装目录的bin目录下执行命令（kafka-topics.sh脚本文件位于kafka安装目录的bin目录下，因此执行kafka-topics.sh需要先进入到bin目录，后面

./kafka-topics.sh --zookeeper localhost:2181 --list

2、查看某个topic的信息

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic-test1

其中topic-test1为需要查看主题的名称。

运用该命令也可以查看多个topic的信息，使用命令：

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic-test1, topic1

注意topic-test1后的逗号与topic1之间不要有空格

3. 找出所有包含配置的主题（只会列出包含了与集群不一样配置的主题）

./kafka-topics.sh --describe --zookeeper localhost:2181 --topics-with-overrides

4、找出所有包含失效副本的分区

此时分区的ISR集合小于AR集合

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic1 --under-replicated-partitions

5、找出所有处于离线状态的分区

可以查找出主题中没有leader副本的分区，这些分区处于离线状态，对外界的生产者和消费者不可用。

./kafka-topics.sh --describe --zookeeper localhost:2181 --topic topic1 --unavailable-partitions

6、创建主题

创建一个分区数为4，副本数为1的主题topic2（副本数不能大于broker数目，由于本地测试的kafka服务端只有1个broker，因此指定这里的relication-factor为1）

./kafka-topics.sh --zookeeper localhost:2181 --create --topic topic2 --partitions 4 --replication-factor 1

查看主题，发现topic2确实已经创建：

进入到kafka服务端的配置文件server.properties（在kafka安装路径的config目录下）中的log.dirs指定的日志路径中，查看：

会生成相应的文件夹。

7. 修改主题的分区个数：

将主题topic2的分区数从4改为5：

./kafka-topics.sh --zookeeper localhost:2181 --alter --topic topic2 --partitions 5

查看topic2的信息，发现分区数已经变成了5：

注意这里的warning信息：改变分区数，会影响消息将要发往的分区。比如消息1，根据Key1经过一定算法计算，得出它应该发送到分区1。但此时增大分区数后，原本将要发往分区1的消息，经过同样的计算被发往分区3。这种影响也会间接的改变分区中消息的顺序。

注意：通过alter更改主题的分区数，只能增大分区数，不能减少分区数。

比如想要将topic2的分区数从5改为4，会报错：

8、删除主题

只有在broker端配置参数delete.topic.enable为true才能够删除主题，使用下面命令删除主题topic2：

./kafka-topics.sh --zookeeper localhost:2181 --delete --topic topic2

9、手动对分区leader副本执行重新平衡

./kafka-preferred-replica-election.sh --zookeeper localhost:2181

10、查看分区副本的日志文件

每个分区的每个副本都会对应着一个唯一的用于存储消息的日志文件，比如创建主题topic-test1，并且分区数为4，副本数为1，则一共会生成四个与topic1相关的日志文件。在kafka日志目录下，可以看到这些.log文件：

cd topic1-0，

其中00000000000000000027.log就是存储消息的日志文件。可以通过kafka提供的脚本进行查看

进入kafka安装目录下的bin目录，执行：

./kafka-dump-log.sh --files /home/zou/kafka/log/topic1-0/00000000000000000027.log --print-data-log

其中/home/zou/kafka/log/topic1-0/00000000000000000027.log是对应日志文件的路径，输出如下：

Dumping /home/zou/kafka/log/topic1-0/00000000000000000027.log
Starting offset: 27
offset: 27 position: 0 CreateTime: 1635939310593 isvalid: true keysize: -1 valuesize: 21 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: prefix1-测试消息1
offset: 28 position: 89 CreateTime: 1635991823796 isvalid: true keysize: -1 valuesize: 21 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: prefix1-测试消息2

执行

./kafka-dump-log.sh --files /home/zou/kafka/log/topic1-0/00000000000000000027.log

结果：

Dumping /home/zou/kafka/log/topic1-0/00000000000000000027.log
Starting offset: 27
baseOffset: 27 lastOffset: 27 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 0 CreateTime: 1635939310593 isvalid: true size: 89 magic: 2 compresscodec: NONE crc: 190681321
baseOffset: 28 lastOffset: 28 count: 1 baseSequence: -1 lastSequence: -1 producerId: -1 producerEpoch: -1 partitionLeaderEpoch: 0 isTransactional: false isControl: false position: 89 CreateTime: 1635991823796 isvalid: true size: 89 magic: 2 compresscodec: NONE crc: 2060523724

11、查看kafka日志的index文件

偏移量索引文件

首先偏移量索引文件（xxx.index文件）结构如下：

xxx：为baseOffset

relativeOffset：相对偏移量，表示消息相对于baseOffset的偏移量

position：为物理位置（消息在日志分段文件logsegment文件中的物理位置）

比如：00000000000000000029.index

进入到kafka安装目录的bin目录下，使用以下命令查看偏移量索引文件：

./kafka-dump-log.sh --files /home/zou/kafka/log/topic1-0/00000000000000000029.index

其中：/home/zou/kafka/log/topic1-0/00000000000000000029.index为index文件的绝对路径

结果如下：

注意index索引文件为稀疏索引，只有1条记录不代表该日志分段文件中只有一条消息。

注意若.index内容为空，运行上述脚本会报异常：

时间戳索引文件：

也可以用上述命令进行查看