【Kafka】动手实践之Kafka运维脚本系列（持续更新）

Love The Way You Lie 2023-10-05 16:21 42阅读 0赞

#### 文章目录 ####

*  kafka-delete-records.sh
 *  kafka-log-dirs.sh
 *  kafka-preferred-replica-election.sh
 *  kafka-dump-log.sh
 *  kafka-producer-perf-test.sh
 *  kafka-reassign-partitions.sh

## kafka-delete-records.sh ##

**作用**：删除topic指定partition的指定offset之前的数据

**用法：**

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>是否必须</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>--bootstrap-server </td> 
   <td>REQUIRED</td> 
   <td>设置连接的Kafka服务地址，多个以英文逗号分隔</td> 
  </tr> 
  <tr> 
   <td>--command-config </td> 
   <td></td> 
   <td>包含需要传递给Admin Client的配置的属性文件</td> 
  </tr> 
  <tr> 
   <td>--help</td> 
   <td></td> 
   <td>打印help信息</td> 
  </tr> 
  <tr> 
   <td>--offset-json-file </td> 
   <td>REQUIRED</td> 
   <td>带有每个分区offset的json文件，文件内容格式为：{ <br> “partitions”:[<br> { <br> “topic”:“foo”,<br> “partition”:1,<br> “offset”:1<br> }<br> ],<br> “version”:1<br>}</td> 
  </tr> 
  <tr> 
   <td>--version</td> 
   <td></td> 
   <td>显示Kafka server版本信息</td> 
  </tr> 
 </tbody> 
</table>

**用法：**

[root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-delete-records.sh --bootstrap-server 172.24.29.163:9092,172.24.29.164:9092,172.24.29.165:9092 --offset-json-file files/offset.json
    Executing records delete operation
    Records delete operation completed:
    partition: zhurunhua-test-1-3   low_watermark: 12
    partition: zhurunhua-test-1-1   low_watermark: 15
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

offset.json文件内容：

{
        
        "partitions":[
            {
        
                "topic":"zhurunhua-test-1",
                "partition":1,
                "offset":15
            },
            {
        
                "topic":"zhurunhua-test-1",
                "partition":3,
                "offset":12
            }
        ],
        "version":1
    }

**删除之前：**

partition 1:![3ba4dc72ee504a5aea8d363c6141f221.png][]partition 3:  
![2b0f02b9afce09a560e5e048ea8a5513.png][]  
**删除之后：**

partition 1:  
![4e656a43072043991be71e2d35cc9555.png][]  
partition 3:  
![8904f128cb5783423379b267693a74c2.png][]  
**总结**：

partition 1的offset为15之前的消息、partition 3的offset为12之前的消息被删了，可以理解为删除每个partition中的消息至指定的低水位(LW)

## kafka-log-dirs.sh ##

**作用**：查看kafka消息日志的目录信息

**用法：**

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>是否必须</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>--bootstrap-server </td> 
   <td>REQUIRED</td> 
   <td>设置连接的Kafka服务地址，多个以英文逗号分隔</td> 
  </tr> 
  <tr> 
   <td>--broker-list </td> 
   <td></td> 
   <td>需要查询的broker的id，多个以逗号分隔，如：0,1,2，不传则查询所有broker</td> 
  </tr> 
  <tr> 
   <td>--command-config </td> 
   <td></td> 
   <td>Admin Client的配置文件</td> 
  </tr> 
  <tr> 
   <td>--describe</td> 
   <td>REQUIRED</td> 
   <td>官方解释：描述指定Broker上的指定日志目录，试了一下，不传不行</td> 
  </tr> 
  <tr> 
   <td>--topic-list </td> 
   <td></td> 
   <td>需要查询的topic，多个以逗号分隔，不传则查询所有topic</td> 
  </tr> 
 </tbody> 
</table>

**示例：**

[root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-log-dirs.sh --bootstrap-server 172.24.29.163:9092,172.24.29.164:9092,172.24.29.165:9092 --topic-list zhurunhua-test-1 --broker-list 1 --describe
    Querying brokers for log directories information
    Received log directory information from brokers 1
    {
        "version":1,"brokers":[{
        "broker":1,"logDirs":[{
        "logDir":"/neworiental/kafka211/kafka-logs","error":null,"partitions":[{
        "partition":"zhurunhua-test-1-4","size":1768,"offsetLag":0,"isFuture":false},{
        "partition":"zhurunhua-test-1-1","size":3436,"offsetLag":0,"isFuture":false},{
        "partition":"zhurunhua-test-1-13","size":3094,"offsetLag":0,"isFuture":false},{
        "partition":"zhurunhua-test-1-10","size":3094,"offsetLag":0,"isFuture":false},{
        "partition":"zhurunhua-test-1-7","size":2431,"offsetLag":0,"isFuture":false}]}]}]}
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

可以通过查询到的结果，进入指定目录验证：

[root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-4]# 
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-4]# ll
    total 12
    -rw-r--r-- 1 root root 10485760 Dec  8 11:41 00000000000000000005.index
    -rw-r--r-- 1 root root     1768 Dec 10 17:29 00000000000000000005.log
    -rw-r--r-- 1 root root       10 Dec  8 11:41 00000000000000000005.snapshot
    -rw-r--r-- 1 root root 10485756 Dec  8 11:41 00000000000000000005.timeindex
    -rw-r--r-- 1 root root        8 Dec  8 11:41 leader-epoch-checkpoint
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-4]# 
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-4]# pwd
    /neworiental/kafka211/kafka-logs/zhurunhua-test-1-4
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-4]# 
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-4]# cd ../zhurunhua-test-1-1
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-1]# 
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-1]# ll
    total 8
    -rw-r--r-- 1 root root 10485760 Dec  8 11:41 00000000000000000007.index
    -rw-r--r-- 1 root root     3436 Dec 10 17:29 00000000000000000007.log
    -rw-r--r-- 1 root root 10485756 Dec  8 11:41 00000000000000000007.timeindex
    -rw-r--r-- 1 root root        9 Dec 10 18:26 leader-epoch-checkpoint
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-1]# 
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-1]# pwd
    /neworiental/kafka211/kafka-logs/zhurunhua-test-1-1
    [root@XXGL-T-TJSYZ-app-025 zhurunhua-test-1-1]#

**总结：**

可以指定查询集群中每个Broker上(指定)Topic的日志存放目录信息，包括：

*  broker：broker id
 *  logDir：该broker上存放该Topic的日志目录
 *  error：
 *  partition：partition名称（每个partition对应一个子目录）
 *  size：该partition上日志文件(xxx.log)大小
 *  offsetLag：与Leader之间复制的滞后？？？
 *  isFuture：

## kafka-preferred-replica-election.sh ##

**作用**：使每个分区的Leader转移回首选副本(preferred replica)，用来平衡集群内部的领导权，即对分区Leader进行重新平衡。

**用法：**

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>是否必须</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>--path-to-json-file </td> 
   <td></td> 
   <td>需要执行优先副本选举的分区清单，设置哪些partition需要触发优先副本选举，格式：<br> {“partitions”: <br> [{“topic”: “foo”, “partition”: 1}, <br> {“topic”: “foobar”, “partition”: 2}]<br> }</td> 
  </tr> 
  <tr> 
   <td>--zookeeper </td> 
   <td>REQUIRED</td> 
   <td>zookeeper地址</td> 
  </tr> 
 </tbody> 
</table>

**注意：**–path-to-json-file参数可以指定partition，执行小批量的优先副本选举操作，如果不指定，会去zookeeper查询所有的partition，对所有Leader不是“preferred replica”的partition做优先副本选举，在实际生产环境中，一般使用path-to-json-file参数来分批、手动地执行优先副本选举操作，尤其是在应对大规模的Kafka集群时，同时，优先副本的选举操作也要注意避开业务高峰期，以免带来性能方面的负面影响。

**示例：**

先看看topic：test\_producer\_9\_3中partition的初始状态信息

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_producer_9_3 --describe
    Topic:test_producer_9_3 PartitionCount:9        ReplicationFactor:3     Configs:retention.ms=60000
            Topic: test_producer_9_3        Partition: 0    Leader: 1       Replicas: 1,2,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 2    Leader: 3       Replicas: 3,1,2 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 3    Leader: 1       Replicas: 1,3,2 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 4    Leader: 2       Replicas: 2,1,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 5    Leader: 3       Replicas: 3,2,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 6    Leader: 1       Replicas: 1,2,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 7    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 8    Leader: 3       Replicas: 3,1,2 Isr: 2,3,1
    [root@XXGL-T-TJSYZ-app-025 bin]#

可以看到，所有partition的的leader都是都是AR集合中的第一个，也就是preferred replica。

接下来把broker 1杀掉

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_producer_9_3 --describe
    Topic:test_producer_9_3 PartitionCount:9        ReplicationFactor:3     Configs:retention.ms=60000
            Topic: test_producer_9_3        Partition: 0    Leader: 2       Replicas: 1,2,3 Isr: 2,3
            Topic: test_producer_9_3        Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 2,3
            Topic: test_producer_9_3        Partition: 2    Leader: 3       Replicas: 3,1,2 Isr: 2,3
            Topic: test_producer_9_3        Partition: 3    Leader: 3       Replicas: 1,3,2 Isr: 2,3
            Topic: test_producer_9_3        Partition: 4    Leader: 2       Replicas: 2,1,3 Isr: 2,3
            Topic: test_producer_9_3        Partition: 5    Leader: 3       Replicas: 3,2,1 Isr: 2,3
            Topic: test_producer_9_3        Partition: 6    Leader: 2       Replicas: 1,2,3 Isr: 2,3
            Topic: test_producer_9_3        Partition: 7    Leader: 2       Replicas: 2,3,1 Isr: 2,3
            Topic: test_producer_9_3        Partition: 8    Leader: 3       Replicas: 3,1,2 Isr: 2,3
    [root@XXGL-T-TJSYZ-app-025 bin]#

可以看到partition：0、3、6的leader由之前的优先副本变成了ISR集合中的第一个：2

启动broker 1之后，查看topic信息

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_producer_9_3 --describe
    Topic:test_producer_9_3 PartitionCount:9        ReplicationFactor:3     Configs:retention.ms=60000
            Topic: test_producer_9_3        Partition: 0    Leader: 2       Replicas: 1,2,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 2    Leader: 3       Replicas: 3,1,2 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 3    Leader: 3       Replicas: 1,3,2 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 4    Leader: 2       Replicas: 2,1,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 5    Leader: 3       Replicas: 3,2,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 6    Leader: 2       Replicas: 1,2,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 7    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 8    Leader: 3       Replicas: 3,1,2 Isr: 2,3,1
    [root@XXGL-T-TJSYZ-app-025 bin]#

可以看到重新启动的broker 1成为了当前leader的follower

现在开始测试脚本kafka-preferred-replica-election.sh

json文件如下：

{
          
     "partitions":  
      [  
        {
        "topic": "test_producer_9_3", "partition": 3},
        {
        "topic": "test_producer_9_3", "partition": 6}
      ]  
    }

执行脚本：

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-preferred-replica-election.sh --zookeeper 172.24.29.163:2181/kafka211 --path-to-json-file files/election.json
    Created preferred replica election path with test_producer_9_3-3,test_producer_9_3-6
    Successfully started preferred replica election for partitions Set(test_producer_9_3-3, test_producer_9_3-6)
    [root@XXGL-T-TJSYZ-app-025 bin]#

再查看一下topic的信息：

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_producer_9_3 --describe
    Topic:test_producer_9_3 PartitionCount:9        ReplicationFactor:3     Configs:retention.ms=60000
            Topic: test_producer_9_3        Partition: 0    Leader: 2       Replicas: 1,2,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 1    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 2    Leader: 3       Replicas: 3,1,2 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 3    Leader: 1       Replicas: 1,3,2 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 4    Leader: 2       Replicas: 2,1,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 5    Leader: 3       Replicas: 3,2,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 6    Leader: 1       Replicas: 1,2,3 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 7    Leader: 2       Replicas: 2,3,1 Isr: 2,3,1
            Topic: test_producer_9_3        Partition: 8    Leader: 3       Replicas: 3,1,2 Isr: 2,3,1
    [root@XXGL-T-TJSYZ-app-025 bin]#

可以看到partition：3、6的leader已经恢复到为broker 1了，而未执行优先副本选举的partition 0的leader还是broker 2

**补充：**

**什么是优先副本(preferred replica)？**

在主题创建的时候，该主题的分区及副本会尽可能均匀地分布到集Kafka群中的各个broker节点上，对应的leader副本的分配也比较均匀，随着时间的更替，Kafka集群中的broker节点不可避免地会遇到宕机或者崩溃的问题，此时，该broker节点上所有角色为leader的partition的会在其他副本重新选举leader，当该节点恢复的时候，对于之前的这些partition，会重新以follower的角色加入到集群中，这就导致了集群中leader分配不均衡，leader的分配不均衡，从一定程度上讲，就会导致集群内所有节点的负载不均衡。

为了有效治理集群负载失衡的情况，Kafka引入了优先副本的概念，所谓优先副本就是指在AR集合列表中的第一个副本。理想情况下，优先副本就是该分区的leader，所以也称为preferred leader。

**什么是优先副本选举(preferred leader election)？**

优先副本选举是指通过一定的方式，促使优先副本选举为leader副本，以此来促进集群的负载均衡，这一行为也可以称为“分区平衡”。但是要注意，就算集群中的分区分配均衡、leader分配均衡，也不能确保整个集群的负载是均衡的，因为每个分区承载的负荷是不一样的。

在Kafka中可以提供分区自动平衡的功能，需要通过broker端的参数：auto.leader.rebalance.enable来设置，此参数的默认值是true，如果开启此功能，Kafka集群的controller会启动一个定时任务，轮询所有的broker节点，计算每个broker节点的分区不平衡率(broker中的不平衡率=leader不是优先副本的分区个数/分区总数)是否超过leader.imbalance.per.broker.percentege参数配置的比值，默认为10%，如果超过设定的比值，会自动执行优先副本选举的动作以求分区平衡。执行周期由参数leader.imbalance.check.interal.seconds控制，默认300秒。

优先副本的选举过程是一个安全的过程，Kafka客户端可以自动感知分区leader副本的变更。

**优先副本会改变吗？**

经过很多次试验，在非人为使用脚本操作的情况下，将集群中的broker节点杀掉、再恢复，发现AR集合是不会改变的，而优先副本是AR集合中的第一个，也就是说，topic的partition一旦创建成功，优先副本是不会改变的。  
当然使用kafka-reassign-partitions.sh是可以改变每个partition的AR集合顺序的，也就是改变partition的优先副本。后续有详细操作示例。

**如果优先副本不在ISR中会怎么样？**  
如果preferred replica不在ISR中，控制器将无法将leader转移到preferred replica。

## kafka-dump-log.sh ##

**作用**：解析日志文件并将其内容输出到控制台。

**用法**：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>是否必须</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>--deep-iteration</td> 
   <td></td> 
   <td>如果设置了，则使用深度而不是浅层迭代，如果设置了–print-data-log，则自动设置改选项</td> 
  </tr> 
  <tr> 
   <td>–files &lt;String: file1, file2, …&gt;</td> 
   <td>REQUIRED</td> 
   <td>文件列表，多个以逗号分隔</td> 
  </tr> 
  <tr> 
   <td>--help</td> 
   <td></td> 
   <td>打印帮助信息</td> 
  </tr> 
  <tr> 
   <td>--index-sanity-check</td> 
   <td></td> 
   <td>如果设置了，只需检查索引的健全性，而不打印其内容</td> 
  </tr> 
  <tr> 
   <td>--key-decoder-class [String]</td> 
   <td></td> 
   <td>设置反序列化key所需要的类，默认：kafka.serializer.StringDecoder</td> 
  </tr> 
  <tr> 
   <td>--max-message-size &lt;Integer: size&gt;</td> 
   <td></td> 
   <td>最大消息的大小，默认：5242880</td> 
  </tr> 
  <tr> 
   <td>--offsets-decoder</td> 
   <td></td> 
   <td>如果设置了，日志数据将根据主题__consumer_offsets解析为偏移量数据</td> 
  </tr> 
  <tr> 
   <td>--print-data-log</td> 
   <td></td> 
   <td>是否打印消息内容，如果设置了任何decoder操作，会自动设置该选项</td> 
  </tr> 
  <tr> 
   <td>--transaction-log-decoder</td> 
   <td></td> 
   <td>如果设置了，日志数据将通过主题__transaction_state解析为事务元数据</td> 
  </tr> 
  <tr> 
   <td>--value-decoder-class [String]</td> 
   <td></td> 
   <td>设置反序列化消息的类，默认：kafka.serializer.StringDecoder</td> 
  </tr> 
  <tr> 
   <td>--verify-index-only</td> 
   <td></td> 
   <td>如果设置了，只验证索引日志，不打印内容</td> 
  </tr> 
 </tbody> 
</table>

**示例**：

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-dump-log.sh --files /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000005961.index
    Dumping /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000005961.index
    offset: 5983 position: 4119
    offset: 6005 position: 8220
    offset: 6027 position: 12336
    offset: 6049 position: 16445
    offset: 6072 position: 20718
    offset: 6094 position: 24827
    offset: 6116 position: 28944
    offset: 6139 position: 33209
    offset: 6161 position: 37326
    offset: 6183 position: 41434
    [root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-dump-log.sh --files /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000005961.index --index-sanity-check
    Dumping /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000005961.index
    /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000005961.index passed sanity check.
    [root@XXGL-T-TJSYZ-app-025 bin]#

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-dump-log.sh --files /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000005961.log --print-data-log
    Dumping /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000005961.log
    Starting offset: 5961
    offset: 5961 position: 0 CreateTime: 1608023345782 isvalid: true keysize: -1 valuesize: 106 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {
        "ip":"172.24.130.113","port":6379,"psd":"fkqg8ebo","sysName":"PayMS项目","redisCode":"redismzdnmshgbs"}
    offset: 5962 position: 176 CreateTime: 1608026945503 isvalid: true keysize: -1 valuesize: 115 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {
        "ip":"172.24.130.177","port":6379,"psd":"eup8gthb","sysName":"[用户]教师中心","redisCode":"redisusneqymmzo"}
    offset: 5963 position: 361 CreateTime: 1608026946064 isvalid: true keysize: -1 valuesize: 136 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {
        "ip":"172.24.131.241","port":6379,"psd":"ehpwjfve","sysName":"[用户]q环境-资源管理-教室工程","redisCode":"redisljz96towgt"}
    offset: 5964 position: 567 CreateTime: 1608030545895 isvalid: true keysize: -1 valuesize: 106 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {
        "ip":"172.24.130.101","port":6379,"psd":"5wcovjk8","sysName":"[用户]hub","redisCode":"redisivpmo7rbeh"}
    offset: 5965 position: 743 CreateTime: 1608034145720 isvalid: true keysize: -1 valuesize: 109 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {
        "ip":"172.24.130.60","port":6379,"psd":"mn7dgvnn","sysName":"arch-middleware","redisCode":"redisqihyfklehr"}
    offset: 5966 position: 922 CreateTime: 1608037745467 isvalid: true keysize: -1 valuesize: 114 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {
        "ip":"172.24.130.32","port":6379,"psd":"jvgitvps","sysName":"[用户]xcloud测试","redisCode":"redisuvdxjsun9l"}
    offset: 5967 position: 1106 CreateTime: 1608037746066 isvalid: true keysize: -1 valuesize: 107 magic: 2 compresscodec: NONE producerId: -1 producerEpoch: -1 sequence: -1 isTransactional: false headerKeys: [] payload: {
        "ip":"172.24.130.174","port":6379,"psd":"65hgez9q","sysName":"用户中心","redisCode":"redis9txz6eyraj"}

**补充**：

**Kafka文件目录布局**

![7b712d2a1a6262c56d81988ae47f95fb.png][]

Kafka的log文件对应了一个命名形式为topic-partition的文件夹，每个分区的文件夹是同级的，下面就展示了topic名为slowlog，在当前broker上的所有partition对应的目录

[root@XXGL-T-TJSYZ-app-025 kafka-logs]# ll |grep slowlog
    drwxr-xr-x 2 root root 4096 Dec 24 14:29 slowlog-0
    drwxr-xr-x 2 root root 4096 Dec 24 14:29 slowlog-12
    drwxr-xr-x 2 root root 4096 Dec 24 14:29 slowlog-15
    drwxr-xr-x 2 root root 4096 Dec 24 14:29 slowlog-3
    drwxr-xr-x 2 root root 4096 Dec 24 14:29 slowlog-6
    drwxr-xr-x 2 root root 4096 Dec 24 14:29 slowlog-9
    [root@XXGL-T-TJSYZ-app-025 kafka-logs]#

进入到其中一个文件夹

[root@XXGL-T-TJSYZ-app-025 kafka-logs]# cd slowlog-0
    [root@XXGL-T-TJSYZ-app-025 slowlog-0]# ll
    total 124
    -rw-r--r-- 1 root root       80 Dec 22 17:09 00000000000000005961.index
    -rw-r--r-- 1 root root    44600 Dec 22 16:09 00000000000000005961.log
    -rw-r--r-- 1 root root      132 Dec 22 17:09 00000000000000005961.timeindex
    -rw-r--r-- 1 root root 10485760 Dec 29 05:09 00000000000000006200.index
    -rw-r--r-- 1 root root    44190 Dec 29 15:09 00000000000000006200.log
    -rw-r--r-- 1 root root       10 Dec 22 17:09 00000000000000006200.snapshot
    -rw-r--r-- 1 root root 10485756 Dec 29 05:09 00000000000000006200.timeindex
    -rw-r--r-- 1 root root       10 Dec 24 11:50 00000000000000006261.snapshot
    -rw-r--r-- 1 root root       10 Dec 24 13:53 00000000000000006264.snapshot
    -rw-r--r-- 1 root root       10 Dec 24 14:13 00000000000000006265.snapshot
    -rw-r--r-- 1 root root       36 Dec 24 14:29 leader-epoch-checkpoint
    [root@XXGL-T-TJSYZ-app-025 slowlog-0]#

最主要的三个文件：日志文件(.log)、偏移量索引文件(.index)、时间戳索引文件(.timeindex)

**日志文件(.log)**

为了防止每个分区的日志文件过大，Kafka引入了日志分段(LogSegment)的概念，将一个Log切分为多个LogSegment，这样便于消息的维护和清理。每一个.log文件称为一个LogSegment，向LogSegment中追加消息时是顺序写入的，只有最后一个LogSegment才能执行写入操作，为了方便此描述，将最后一个LogSegment称为“activeSegment”，即表示当前活跃的日志分段，随着消息的不断写入，当activeSegment满足一定条件时，需要创建新的activeSegment，之后追加的消息将写入新的activeSegment。

每个 LogSegment 都有一个基准偏移量(baseOffset)，用来表示当前LogSegment中第一条消息的offset。偏移量是一个64位的长整型数，固定是20位数字，长度未达到，用0进行填补，索引文件和日志文件都由该作为文件名命名规则。比如，如果上述有一个日志文件名为：00000000000000005961.log ，代表当前日志文件的一条数据偏移量就是5961。

**偏移量索引文件(.index)**

作用：通过索引查询消息

偏移量索引项的格式如下图，每个索引项(每行数据)占8个字节，分为两个部分：

![89ae5b802426b10044ae1704e6df6fa0.png][]

*  relativeOffset：相对偏移量，表示消息相对于baseOffset的偏移量，也就是相对于当前日志段中第一条消息的偏移量，占用4个字节；
 *  position：物理地址，也就是消息在日志分段中对应的物理地址，占用4个字节，我理解的是在磁盘上存储该条消息的具体槽位地址。

消息的偏移量(offset)占用8个字节，也可以成为绝对偏移量，索引项中使用的相对偏移量只占用4个字节，这样可以减少索引文件占用的空间(relativeOffset=offset-baseOffset)。

**时间戳索引文件(.timeindex)**

作用：通过时间查询消息

时间戳索引项格式如下：

![9d12852b9e92933c0e7b90c693574855.png][]

每个索引项占12个字节，分为两个部分：

*  timestamp：当前日志分段的最大时间戳；
 *  relativeOffset：时间戳所对应消息的相对偏移量

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-dump-log.sh --files /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000006200.timeindex 
    Dumping /neworiental/kafka211/kafka-logs/slowlog-0/00000000000000006200.timeindex
    timestamp: 1608682150929 offset: 6222
    timestamp: 1608743345795 offset: 6245
    timestamp: 1608779345846 offset: 6260
    timestamp: 1608786546105 offset: 6263
    timestamp: 1608790145815 offset: 6264
    timestamp: 1608847750804 offset: 6287
    timestamp: 1608908945671 offset: 6310
    timestamp: 1608962945910 offset: 6332
    timestamp: 1609016946094 offset: 6354
    timestamp: 1609078145880 offset: 6377
    timestamp: 1609132146003 offset: 6399
    timestamp: 1609189745470 offset: 6421
    Found timestamp mismatch in :/neworiental/kafka211/kafka-logs/slowlog-0/00000000000000006200.timeindex
      Index timestamp: 0, log timestamp: 1608628145850
    Found out of order timestamp in :/neworiental/kafka211/kafka-logs/slowlog-0/00000000000000006200.timeindex
      Index timestamp: 0, Previously indexed timestamp: 1609189745470
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

每个追加的时间戳索引项中的timestamp必须大于之前追加的索引项的timestamp，否则不予以追加，如果broker端参数log.message.timstamp.type设置为LogAppendTime，那么消息的时间戳必定能保持单调递增，如果设置为CreateTime则无法保证，生产者发送消息时，可以通过参数指定消息的时间戳，如果两个不同时钟的生产者同时往一个分区中写入消息，那么会造成当前分区的时间戳乱序。

每当写入一定量的消息时，就会在.index文件和.timeindex文件中分别增加一个偏移量索引项和时间戳索引项，两个文件增加索引项的操作是同时进行的，但不意味着两个索引项对应的relativeOffset是一样的。

**切分文件**

当满足如下几个条件中的其中之一，就会触发文件的切分：

1.  当前日志分段文件的大小超过了 broker 端参数 `log.segment.bytes` 配置的值。`log.segment.bytes` 参数的默认值为 1073741824，即 1GB。
2.  当前日志分段中消息的最早时间戳与当前系统的时间戳的差值大于 `log.roll.ms` 或 `log.roll.hours` 参数配置的值。如果同时配置了 `log.roll.ms` 和 `log.roll.hours` 参数，那么 `log.roll.ms` 的优先级高。默认情况下，只配置了 `log.roll.hours` 参数，其值为168，即 7 天。
3.  偏移量索引文件或时间戳索引文件的大小达到 broker 端参数 `log.index.size.max.bytes` 配置的值。`log.index.size.max.bytes` 的默认值为 10485760，即 10MB。
4.  追加的消息的偏移量与当前日志分段的偏移量之间的差值大于 `Integer.MAX_VALUE`，即要追加的消息的偏移量不能转变为相对偏移量。

**为什么是 `Integer.MAX_VALUE` ？**

在偏移量索引文件中，每个索引项共占用 8 个字节，并分为两部分。相对偏移量和物理地址，每个部分4个字节，4个字节刚好对应 `Integer.MAX_VALUE` ，如果大于 `Integer.MAX_VALUE` ，则不能用 4 个字节进行表示了。

## kafka-producer-perf-test.sh ##

**作用**：用来验证/测试producer的性能

**用法**：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>是否必须</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>-h, --help</td> 
   <td></td> 
   <td>显示帮助信息</td> 
  </tr> 
  <tr> 
   <td>--topic </td> 
   <td>REQUIRED</td> 
   <td>设置topic名称</td> 
  </tr> 
  <tr> 
   <td>--num-records </td> 
   <td>REQUIRED</td> 
   <td>需要生产的消息数</td> 
  </tr> 
  <tr> 
   <td>--payload-delimiter </td> 
   <td></td> 
   <td>文件中的消息分隔符，只有设置了参数–payload-file时才有效，如果没设置，会忽略改参数，默认为\n，即换行符</td> 
  </tr> 
  <tr> 
   <td>--throughput </td> 
   <td>REQUIRED</td> 
   <td>限制消息生产的吞吐量(大约，不是精确限制)，即限制每秒最多发送多少条消息，设置为-1表示不做限制</td> 
  </tr> 
  <tr> 
   <td>--producer-props </td> 
   <td></td> 
   <td>设置生产者的配置参数，格式 key1=value1 key2=value2</td> 
  </tr> 
  <tr> 
   <td>--producer.config &lt;String: file-path&gt;</td> 
   <td></td> 
   <td>生产者的配置参数，以文件的形式提供</td> 
  </tr> 
  <tr> 
   <td>--print-metrics</td> 
   <td></td> 
   <td>是否在测试的最后，打印各项指标信息</td> 
  </tr> 
  <tr> 
   <td>--transactional-id </td> 
   <td></td> 
   <td>事务ID，只有–transaction-duration-ms的值大于0才有效，默认：performance-producer-default-transactional-id</td> 
  </tr> 
  <tr> 
   <td>--transaction-duration-ms &lt;int: positive&gt;</td> 
   <td></td> 
   <td>每个事务的最大期限，超过这个时间，将会提交当前事务，值必须是正数</td> 
  </tr> 
  <tr> 
   <td>--record-size </td> 
   <td></td> 
   <td>每条消息的大小（单位：bytes）</td> 
  </tr> 
  <tr> 
   <td>–payload-file &lt;String: file-path&gt;</td> 
   <td></td> 
   <td>通过文件指定发送的消息内容，文件以UTF-8编码，发送的时候从文件中随机选择一条发送(需结合–payload-delimiter使用)</td> 
  </tr> 
 </tbody> 
</table>

**注意**：

*  –producer-props和–producer.config必须指定其中一个，如果都指定了，优先选择–producer-props配置的；
 *  –record-size 和 --payload-file 必须指定其中一个，而且不能同时指定，否则会报错：producer-performance: error: argument --record-size: not allowed with argument --payload-file

**示例**：

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-producer-perf-test.sh --throughput -1 --topic slowlog --producer-props bootstrap.servers=172.24.29.165:9092 acks=1 --payload-file ./files/messages.txt --num-records 100
    Reading payloads from: /neworiental/kafka211/kafka_2.12-2.1.1/bin/./files/messages.txt
    Number of messages read: 11
    100 records sent, 471.698113 records/sec (0.00 MB/sec), 19.77 ms avg latency, 199.00 ms max latency, 18 ms 50th, 20 ms 95th, 199 ms 99th, 199 ms 99.9th.
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

messages.txt文件内容

[root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]# cat files/messages.txt 
    test-1
    test-2
    test-3
    test-4
    test-5
    test-6
    test-7
    test-8
    test-9
    test-0
    test-11
    [root@XXGL-T-TJSYZ-app-025 bin]#

也可以不指定文件，直接指定消息的大小，由测试程序自动生成消息

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-producer-perf-test.sh --throughput -1 --topic slowlog --producer-props bootstrap.servers=172.24.29.165:9092 acks=1 --record-size 100 --num-records 100               
    100 records sent, 485.436893 records/sec (0.05 MB/sec), 17.83 ms avg latency, 196.00 ms max latency, 16 ms 50th, 18 ms 95th, 196 ms 99th, 196 ms 99.9th.
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

## kafka-reassign-partitions.sh ##

**作用**：用来做集群中broker上线、下线时的分区重分配，以及数据迁移等。

**用法**：

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>是否必须</th> 
   <th>备注</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>--bootstrap-server </td> 
   <td></td> 
   <td>Kafka服务端地址，当--reassignment-json-file所指定的json文件中，任意一个副本的log_dirs配置了绝对路径时，该项为必需项</td> 
  </tr> 
  <tr> 
   <td>--broker-list &lt;String: brokerlist&gt;</td> 
   <td></td> 
   <td>在使用--generate和--topics-to-move-json-file生成重分配配置时，该项为必需项，可指定对哪些broker进行分区重分配，例如：0,1,2</td> 
  </tr> 
  <tr> 
   <td>--command-config &lt;String: file-path&gt;</td> 
   <td></td> 
   <td>Admin client配置文件路径</td> 
  </tr> 
  <tr> 
   <td>--disable-rack-aware</td> 
   <td></td> 
   <td>禁用机架感知副本分配</td> 
  </tr> 
  <tr> 
   <td>--execute</td> 
   <td></td> 
   <td>根据--reassignment-json-file指定的配置，开始执行分配</td> 
  </tr> 
  <tr> 
   <td>--generate</td> 
   <td></td> 
   <td>生成候选分区分配配置（json），只生成，不执行</td> 
  </tr> 
  <tr> 
   <td>--reassignment-json-file &lt;String: file-path&gt;</td> 
   <td></td> 
   <td>分区重分配配置的json文件，格式：<br> {“partitions”: <br> [{“topic”: “foo”, <br> “partition”: 1, <br> “replicas”: [1,2,3], <br> “log_dirs”: [“dir1”,“dir2”,“dir3”] <br> }], <br> “version”:1 <br> } <br>请注意，“log_dirs”是可选的。指定时，其列表长度必须等于replicas列表的长度。此列表中的值可以是“any”，也可以是broker上日志目录的绝对路径。如果指定了绝对日志目录路径，则要求尚未在该broker上创建副本。稍后将在broker上的指定日志目录中创建副本(即指定了绝对路径，代表是扩容之后的重分配操作)。</td> 
  </tr> 
  <tr> 
   <td>--replica-alter-log-dirs-throttle </td> 
   <td></td> 
   <td>顾名思义，限制副本更改日志目录时数据移动的速率阈值，即限制同一个broker上，不同日志目录间，数据移动的速率，单位：bytes/sec，默认：-1</td> 
  </tr> 
  <tr> 
   <td>--throttle </td> 
   <td></td> 
   <td>设置partition数据在broker之间移动的速率阈值，单位：bytes/sec，默认：-1</td> 
  </tr> 
  <tr> 
   <td>--timeout </td> 
   <td></td> 
   <td>等待分区重新分配执行成功启动所允许的最长时间，单位：毫秒，默认：10000</td> 
  </tr> 
  <tr> 
   <td>--topics-to-move-json-file &lt;String: file-path&gt;</td> 
   <td></td> 
   <td>设置针对哪些topic进行分区重分配，这些topic的partition将被移动到–broker-list指定的broker中，文件格式：<br> {“topics”: <br> [{“topic”: “foo”},{“topic”: “foo1”}],<br> “version”:1 <br> }</td> 
  </tr> 
  <tr> 
   <td>--verify</td> 
   <td></td> 
   <td>验证重分配结果</td> 
  </tr> 
  <tr> 
   <td>--zookeeper </td> 
   <td>REQUIRED</td> 
   <td>zookeeper地址</td> 
  </tr> 
 </tbody> 
</table>

**Tips**：kafka-reassign-partitions.sh脚本对应三个操作：生成重分配策略(--generate)、执行重分配(--execute)、验证(--verify)，每个操作结合不用的参数

**示例1**：下线broker时，进行重分配

重分配之前的topic状态

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_reassign --describe    
    Topic:test_reassign     PartitionCount:6        ReplicationFactor:2     Configs:
            Topic: test_reassign    Partition: 0    Leader: 1       Replicas: 1,3   Isr: 1,3
            Topic: test_reassign    Partition: 1    Leader: 2       Replicas: 2,1   Isr: 2,1
            Topic: test_reassign    Partition: 2    Leader: 3       Replicas: 3,2   Isr: 3,2
            Topic: test_reassign    Partition: 3    Leader: 1       Replicas: 1,2   Isr: 1,2
            Topic: test_reassign    Partition: 4    Leader: 2       Replicas: 2,3   Isr: 2,3
            Topic: test_reassign    Partition: 5    Leader: 3       Replicas: 3,1   Isr: 3,1
    [root@XXGL-T-TJSYZ-app-025 bin]#

现在需要下线broker1，生成重分配策略（注意--broker-list指定的是2,3，代表topic下的所有partition只在2，3上做重分配，即排除1）

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --generate --topics-to-move-json-file ./files/topics.json --broker-list 2,3
    Current partition replica assignment
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":1,"replicas":[2,1],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":4,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[1,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[3,1],"log_dirs":["any","any"]}]}
    
    Proposed partition reassignment configuration
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[2,3],"log_dirs":["any","any"]}]}
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

topics.json文件：

{
        
        "topics":[
            {
        
                "topic":"test_reassign"
            }
        ],
        "version":1
    }

将生成的重分配建议(Proposed partition reassignment configuration)对应的json复制到文件中，执行重分配

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --execute --reassignment-json-file ./files/reassign.json 
    Current partition replica assignment
    
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":1,"replicas":[2,1],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[1,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":4,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[1,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[3,1],"log_dirs":["any","any"]}]}
    
    Save this to use as the --reassignment-json-file option during rollback
    Successfully started reassignment of partitions.
    [root@XXGL-T-TJSYZ-app-025 bin]#

执行完之后，会把重分配之前的partition分配信息打印出来，以便回滚

如果要验证查看分区重分配的进度，只需将上面的execute替换为verify即可

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --verify --reassignment-json-file ./files/reassign.json          
    Status of partition reassignment: 
    Reassignment of partition test_reassign-1 completed successfully
    Reassignment of partition test_reassign-3 completed successfully
    Reassignment of partition test_reassign-2 completed successfully
    Reassignment of partition test_reassign-4 completed successfully
    Reassignment of partition test_reassign-0 completed successfully
    Reassignment of partition test_reassign-5 completed successfully
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

再看看重分配之后的topic信息

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_reassign --describe
    Topic:test_reassign     PartitionCount:6        ReplicationFactor:2     Configs:
            Topic: test_reassign    Partition: 0    Leader: 3       Replicas: 3,2   Isr: 3,2
            Topic: test_reassign    Partition: 1    Leader: 2       Replicas: 2,3   Isr: 2,3
            Topic: test_reassign    Partition: 2    Leader: 3       Replicas: 3,2   Isr: 3,2
            Topic: test_reassign    Partition: 3    Leader: 2       Replicas: 2,3   Isr: 2,3
            Topic: test_reassign    Partition: 4    Leader: 2       Replicas: 3,2   Isr: 2,3
            Topic: test_reassign    Partition: 5    Leader: 3       Replicas: 2,3   Isr: 3,2
    [root@XXGL-T-TJSYZ-app-025 bin]#

可以看到，broker1已经不在每个partition的副本中了，此时就可以安全下线broker1了

**示例2**：集群上线新的broker

其实和示例1方法一样，只不过在生成重分配策略时，–broker-list指定的broker列表中，加入新的broker id

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --generate --topics-to-move-json-file ./files/topics.json --broker-list 1,2,3 
    Current partition replica assignment
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[2,3],"log_dirs":["any","any"]}]}
    
    Proposed partition reassignment configuration
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[1,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[1,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,1],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[3,2],"log_dirs":["any","any"]}]}
    [root@XXGL-T-TJSYZ-app-025 bin]#

之后的操作和示例1一样。

注意，如果集群的broker数小于topic设置的副本因子，在执行该脚本生成重分配策略时，是不会成功的

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --generate --topics-to-move-json-file ./files/topics.json --broker-list 2,3  
    Partitions reassignment failed due to Replication factor: 3 larger than available brokers: 2.
    org.apache.kafka.common.errors.InvalidReplicationFactorException: Replication factor: 3 larger than available brokers: 2.

该topic.json中设置的topic副本因子为3，要在两个broker上执行重分配，两者就冲突了。

**示例3**：修改日志目录

先看待执行重分配的json文件，所有副本1对应的日志目录我都设置为新的日志目录/neworiental/kafka211/logs

{
        
        "version":1,
        "partitions":[
            {
        
                "topic":"test_reassign",
                "partition":4,
                "replicas":[
                    2,
                    1
                ],
                "log_dirs":[
                    "any",
                    "/neworiental/kafka211/logs"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":1,
                "replicas":[
                    2,
                    3
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":3,
                "replicas":[
                    1,
                    3
                ],
                "log_dirs":[
                    "/neworiental/kafka211/logs",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":0,
                "replicas":[
                    1,
                    2
                ],
                "log_dirs":[
                    "/neworiental/kafka211/logs",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":2,
                "replicas":[
                    3,
                    1
                ],
                "log_dirs":[
                    "any",
                    "/neworiental/kafka211/logs"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":5,
                "replicas":[
                    3,
                    2
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            }
        ]
    }

如果按之前的脚本执行，不加参数--bootstrap-server，会报错：bootstrap-server needs to be provided in order to reassign replica to the specified log directory

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --execute --reassignment-json-file ./files/reassign-with-new-log-dir.json                
    Current partition replica assignment
    
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[2,3],"log_dirs":["any","any"]}]}
    
    Save this to use as the --reassignment-json-file option during rollback
    Partitions reassignment failed due to bootstrap-server needs to be provided in order to reassign replica to the specified log directory
    kafka.common.AdminCommandFailedException: bootstrap-server needs to be provided in order to reassign replica to the specified log directory
            at kafka.admin.ReassignPartitionsCommand.reassignPartitions(ReassignPartitionsCommand.scala:630)
            at kafka.admin.ReassignPartitionsCommand$.executeAssignment(ReassignPartitionsCommand.scala:221)
            at kafka.admin.ReassignPartitionsCommand$.executeAssignment(ReassignPartitionsCommand.scala:205)
            at kafka.admin.ReassignPartitionsCommand$.main(ReassignPartitionsCommand.scala:65)
            at kafka.admin.ReassignPartitionsCommand.main(ReassignPartitionsCommand.scala)
    
    [root@XXGL-T-TJSYZ-app-025 bin]#

加上--bootstrap-server后，重分配成功

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --bootstrap-server 172.24.29.165:9092 --execute --reassignment-json-file ./files/reassign-with-new-log-dir.json 
    Current partition replica assignment
    
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":4,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[3,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[2,3],"log_dirs":["any","any"]}]}
    
    Save this to use as the --reassignment-json-file option during rollback
    Successfully started reassignment of partitions.
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

进入指定的新目录，可以看到对应partition：0、2、3、4的数据已经迁移过来了，之前的目录里的数据没了

[root@XXGL-T-TJSYZ-app-025 logs]# ll
    total 16
    -rw-r--r-- 1 root root   0 Jan  5 18:05 cleaner-offset-checkpoint
    -rw-r--r-- 1 root root   4 Jan  5 18:06 log-start-offset-checkpoint
    -rw-r--r-- 1 root root  54 Jan  5 18:05 meta.properties
    -rw-r--r-- 1 root root  76 Jan  5 18:06 recovery-point-offset-checkpoint
    -rw-r--r-- 1 root root  76 Jan  5 18:07 replication-offset-checkpoint
    drwxr-xr-x 2 root root 141 Jan  5 18:06 test_reassign-0
    drwxr-xr-x 2 root root 141 Jan  5 18:06 test_reassign-2
    drwxr-xr-x 2 root root 141 Jan  5 18:06 test_reassign-3
    drwxr-xr-x 2 root root 141 Jan  5 18:06 test_reassign-4
    [root@XXGL-T-TJSYZ-app-025 logs]# pwd
    /neworiental/kafka211/logs
    [root@XXGL-T-TJSYZ-app-025 logs]#

[root@XXGL-T-TJSYZ-app-025 kafka-logs]# cd test_reassign-0
    -bash: cd: test_reassign-0: No such file or directory
    [root@XXGL-T-TJSYZ-app-025 kafka-logs]# 
    [root@XXGL-T-TJSYZ-app-025 kafka-logs]# 
    [root@XXGL-T-TJSYZ-app-025 kafka-logs]# ll |grep test_reassign
    [root@XXGL-T-TJSYZ-app-025 kafka-logs]# 
    [root@XXGL-T-TJSYZ-app-025 kafka-logs]# pwd
    /neworiental/kafka211/kafka-logs
    [root@XXGL-T-TJSYZ-app-025 kafka-logs]#

注意：在不同目录之间迁移数据时，目录必须在配置在server.properties的log.dirs中有配置。

**示例4**：是否可以修改partition的优先副本？

先看当前的topic信息：partition 0～5的优先副本分别为：1、2、3、1、2、3

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_reassign --describe    
    Topic:test_reassign     PartitionCount:6        ReplicationFactor:2     Configs:
            Topic: test_reassign    Partition: 0    Leader: 1       Replicas: 1,2   Isr: 2,1
            Topic: test_reassign    Partition: 1    Leader: 2       Replicas: 2,3   Isr: 2,3
            Topic: test_reassign    Partition: 2    Leader: 3       Replicas: 3,1   Isr: 3,1
            Topic: test_reassign    Partition: 3    Leader: 1       Replicas: 1,3   Isr: 3,1
            Topic: test_reassign    Partition: 4    Leader: 2       Replicas: 2,1   Isr: 2,1
            Topic: test_reassign    Partition: 5    Leader: 3       Replicas: 3,2   Isr: 3,2
    [root@XXGL-T-TJSYZ-app-025 bin]#

手动设置重分配策略，将所有partition的AR集合顺序调换

{
        
        "version":1,
        "partitions":[
            {
        
                "topic":"test_reassign",
                "partition":4,
                "replicas":[
                    1,
                    2
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":1,
                "replicas":[
                    3,
                    2
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":3,
                "replicas":[
                    3,
                    1
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":0,
                "replicas":[
                    2,
                    1
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":2,
                "replicas":[
                    1,
                    3
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            },
            {
        
                "topic":"test_reassign",
                "partition":5,
                "replicas":[
                    2,
                    3
                ],
                "log_dirs":[
                    "any",
                    "any"
                ]
            }
        ]
    }

执行重分配

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-reassign-partitions.sh --zookeeper 172.24.29.163:2181/kafka211 --execute --reassignment-json-file ./files/reassign-new-ar.json 
    Current partition replica assignment
    
    {
        "version":1,"partitions":[{
        "topic":"test_reassign","partition":1,"replicas":[2,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":3,"replicas":[1,3],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":2,"replicas":[3,1],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":4,"replicas":[2,1],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":0,"replicas":[1,2],"log_dirs":["any","any"]},{
        "topic":"test_reassign","partition":5,"replicas":[3,2],"log_dirs":["any","any"]}]}
    
    Save this to use as the --reassignment-json-file option during rollback
    Successfully started reassignment of partitions.
    [root@XXGL-T-TJSYZ-app-025 bin]#

执行完之后，所有partition的AR集合都改变了

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_reassign --describe
    Topic:test_reassign     PartitionCount:6        ReplicationFactor:2     Configs:
            Topic: test_reassign    Partition: 0    Leader: 1       Replicas: 2,1   Isr: 2,1
            Topic: test_reassign    Partition: 1    Leader: 2       Replicas: 3,2   Isr: 2,3
            Topic: test_reassign    Partition: 2    Leader: 3       Replicas: 1,3   Isr: 3,1
            Topic: test_reassign    Partition: 3    Leader: 1       Replicas: 3,1   Isr: 3,1
            Topic: test_reassign    Partition: 4    Leader: 2       Replicas: 1,2   Isr: 2,1
            Topic: test_reassign    Partition: 5    Leader: 3       Replicas: 2,3   Isr: 3,2
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

根据优先副本的定义，AR集合的第一个副本即为优先副本，执行一下优先副本选举，看Leader的情况

election.json文件内容：

[root@XXGL-T-TJSYZ-app-025 files]# cat election.json 
    {
          
     "partitions":  
      [  
        {
        "topic": "test_reassign", "partition": 0},
        {
        "topic": "test_reassign", "partition": 1},
        {
        "topic": "test_reassign", "partition": 2},
        {
        "topic": "test_reassign", "partition": 3}
      ]  
    }
    [root@XXGL-T-TJSYZ-app-025 files]#

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-preferred-replica-election.sh --zookeeper 172.24.29.163:2181/kafka211 --path-to-json-file files/election.json
    Created preferred replica election path with test_reassign-0,test_reassign-1,test_reassign-2,test_reassign-3
    Successfully started preferred replica election for partitions Set(test_reassign-0, test_reassign-1, test_reassign-2, test_reassign-3)
    [root@XXGL-T-TJSYZ-app-025 bin]#

[root@XXGL-T-TJSYZ-app-025 bin]# ./kafka-topics.sh --zookeeper 172.24.29.163:2181/kafka211 --topic test_reassign --describe
    Topic:test_reassign     PartitionCount:6        ReplicationFactor:2     Configs:
            Topic: test_reassign    Partition: 0    Leader: 2       Replicas: 2,1   Isr: 2,1
            Topic: test_reassign    Partition: 1    Leader: 3       Replicas: 3,2   Isr: 2,3
            Topic: test_reassign    Partition: 2    Leader: 1       Replicas: 1,3   Isr: 3,1
            Topic: test_reassign    Partition: 3    Leader: 3       Replicas: 3,1   Isr: 3,1
            Topic: test_reassign    Partition: 4    Leader: 2       Replicas: 1,2   Isr: 2,1
            Topic: test_reassign    Partition: 5    Leader: 3       Replicas: 2,3   Isr: 3,2
    [root@XXGL-T-TJSYZ-app-025 bin]# 
    [root@XXGL-T-TJSYZ-app-025 bin]#

可以看到，指定执行优先副本选举的partition 0、1、2、3对应的Leader已经重新选举成了2、3、1、3，所以kafka-reassign-partitions.sh是可以改变优先副本的。

**Tips**：

*  分区重分配是后台执行的，--execute只是开始执行，使用--verify可查看进度和结果；
 *  分区重分配对集群性能有很大的影响，在实际操作中，尽量降低重分配的粒度，分成多个小批次来执行；
 *  如果要将某个broker下线，那么在执行分区重分配动作之前最好先关闭或重启该broker，这样该broker就不再是任何分区的leader了，这样可以减少broker间的流量复制，提升重分配的性能，减少对集群的影响；
 *  在执行完重分配之后，集群内部的Leader不一定是均衡的，需要结合kafka-preferred-replica-election.sh脚本来优化集群负载。

[3ba4dc72ee504a5aea8d363c6141f221.png]: https://img-blog.csdnimg.cn/img_convert/3ba4dc72ee504a5aea8d363c6141f221.png
[2b0f02b9afce09a560e5e048ea8a5513.png]: https://img-blog.csdnimg.cn/img_convert/2b0f02b9afce09a560e5e048ea8a5513.png
[4e656a43072043991be71e2d35cc9555.png]: https://img-blog.csdnimg.cn/img_convert/4e656a43072043991be71e2d35cc9555.png
[8904f128cb5783423379b267693a74c2.png]: https://img-blog.csdnimg.cn/img_convert/8904f128cb5783423379b267693a74c2.png
[7b712d2a1a6262c56d81988ae47f95fb.png]: https://img-blog.csdnimg.cn/img_convert/7b712d2a1a6262c56d81988ae47f95fb.png
[89ae5b802426b10044ae1704e6df6fa0.png]: https://img-blog.csdnimg.cn/img_convert/89ae5b802426b10044ae1704e6df6fa0.png
[9d12852b9e92933c0e7b90c693574855.png]: https://img-blog.csdnimg.cn/img_convert/9d12852b9e92933c0e7b90c693574855.png