【kafka运维】Kafka全网最全最详细运维命令合集(精品强烈建议收藏!!!)

短命女 2022-10-10 01:27 893阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center]

[超强!!! Kafka学习大全!!!(快收藏)][Kafka]

### 文章目录 ###

*   *  1.TopicCommand
     *   *  1.1.Topic创建
         *  1.2.删除Topic
         *  1.3.Topic分区扩容
         *  1.4.查询Topic描述
         *  5.查询Topic列表
     *  2.ConfigCommand
     *   *  2.1 查询配置
         *   *  Topic配置查询
             *  其他配置/clients/users/brokers/broker-loggers 的查询
             *  查询kafka版本信息
         *  2.2 增删改 配置 \`--alter\`
         *   *  Topic添加/修改动态配置
             *  Topic删除动态配置
             *  添加/删除配置同时执行
             *  其他配置同理,只需要类型改下\`--entity-type\`
         *  默认配置
     *  3.副本扩缩、分区迁移、跨路径迁移 kafka-reassign-partitions
     *  4.Topic的发送kafka-console-producer.sh
     *  5. Topic的消费kafka-console-consumer.sh
     *  6.kafka-leader-election Leader重新选举
     *  7. 持续批量推送消息kafka-verifiable-producer.sh
     *  8. 持续批量拉取消息kafka-verifiable-consumer
     *  9.生产者压力测试kafka-producer-perf-test.sh
     *  10.消费者压力测试kafka-consumer-perf-test.sh
     *  11.删除指定分区的消息kafka-delete-records.sh
     *  12. 查看Broker磁盘信息kafka-log-dirs.sh
     *  12. 消费者组管理 kafka-consumer-groups.sh
     *   *  1. 查看消费者列表\`--list\`
         *  2. 查看消费者组详情\`--describe\`
         *  3. 删除消费者组\`--delete\`
         *  4. 重置消费组的偏移量 \`--reset-offsets\`
         *  5. 删除偏移量\`delete-offsets\`
     *  13.查看日志文件 kafka-dump-log.sh
     *  附件
     *  More

> **日常运维**  
> **问题排查**  
> 怎么能够少了滴滴开源的  
> [滴滴开源LogiKM一站式Kafka监控与管控平台][LogiKM_Kafka]

本文所有命令,博主均全部操作验证过,保证准确性; 非复制粘贴拼凑文章; 如果想了解更多工具命令,可在评论区留下评论,博主会择期加上;

**以下大部分运维操作,都可以使用 [LogI-Kafka-Manager][] 在平台上可视化操作;**

## 1.TopicCommand ##

### 1.1.Topic创建 ###

> `bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic test`

--------------------

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--bootstrap-server</code> 指定kafka服务</td> 
   <td>指定连接到的kafka服务; 如果有这个参数,则 <code>--zookeeper</code>可以不需要</td> 
   <td>–bootstrap-server localhost:9092</td> 
  </tr> 
  <tr> 
   <td><code>--zookeeper</code></td> 
   <td>弃用, 通过zk的连接方式连接到kafka集群;</td> 
   <td>–zookeeper localhost:2181 或者localhost:2181/kafka</td> 
  </tr> 
  <tr> 
   <td><code>--replication-factor</code></td> 
   <td>副本数量,注意不能大于broker数量;如果不提供,则会用集群中默认配置</td> 
   <td>–replication-factor 3</td> 
  </tr> 
  <tr> 
   <td><code>--partitions</code></td> 
   <td>分区数量,当创建或者修改topic的时候,用这个来指定分区数;如果创建的时候没有提供参数,则用集群中默认值; 注意如果是修改的时候,分区比之前小会有问题</td> 
   <td>–partitions 3</td> 
  </tr> 
  <tr> 
   <td><code>--replica-assignment</code></td> 
   <td>副本分区分配方式;创建topic的时候可以自己指定副本分配情况;</td> 
   <td><code>--replica-assignment</code> BrokerId-0:BrokerId-1:BrokerId-2,BrokerId-1:BrokerId-2:BrokerId-0,BrokerId-2:BrokerId-1:BrokerId-0 ; 这个意思是有三个分区和三个副本,对应分配的Broker; 逗号隔开标识分区;冒号隔开表示副本</td> 
  </tr> 
  <tr> 
   <td><code>--config</code>&lt;String: name=value&gt;</td> 
   <td>用来设置topic级别的配置以覆盖默认配置;<strong>只在–create 和–bootstrap-server 同时使用时候生效</strong>; 可以配置的参数列表请看文末附件</td> 
   <td>例如覆盖两个配置 <code>--config retention.bytes=123455 --config retention.ms=600001</code></td> 
  </tr> 
  <tr> 
   <td><code>--command-config</code> &lt;String: command 文件路径&gt;</td> 
   <td>用来配置客户端Admin Client启动配置,<strong>只在–bootstrap-server 同时使用时候生效</strong>;</td> 
   <td>例如:设置请求的超时时间 <code>--command-config config/producer.proterties</code>; 然后在文件中配置 request.timeout.ms=300000</td> 
  </tr> 
 </tbody> 
</table>

### 1.2.删除Topic ###

> `bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test`

--------------------

支持正则表达式匹配Topic来进行删除,只需要将topic 用双引号包裹起来  
**例如: 删除以`create_topic_byhand_zk`为开头的topic;**

> > bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic “create\_topic\_byhand\_zk.\*”  
> > `.`表示任意匹配除换行符 \\n 之外的任何单字符。要匹配 . ，请使用 . 。  
> > `·*·`：匹配前面的子表达式零次或多次。要匹配 \* 字符，请使用 \*。  
> > `.*` : 任意字符

**删除任意Topic (慎用)**

> bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic “.\*?”

更多的用法请[参考正则表达式][Link 1]

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--replica-assignment</code></td> 
   <td>副本分区分配方式;创建topic的时候可以自己指定副本分配情况;</td> 
   <td><code>--replica-assignment</code> BrokerId-0:BrokerId-1:BrokerId-2,BrokerId-1:BrokerId-2:BrokerId-0,BrokerId-2:BrokerId-1:BrokerId-0 ; 这个意思是有三个分区和三个副本,对应分配的Broker; 逗号隔开标识分区;冒号隔开表示副本</td> 
  </tr> 
 </tbody> 
</table>

**PS: 虽然这里配置的是全部的分区副本分配配置,但是正在生效的是新增的分区;**  
比如: 以前3分区1副本是这样的

<table> 
 <thead> 
  <tr> 
   <th>Broker-1</th> 
   <th>Broker-2</th> 
   <th>Broker-3</th> 
   <th>Broker-4</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>0</td> 
   <td>1</td> 
   <td>2</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

现在新增一个分区,`--replica-assignment` 2,1,3,4 ; 看这个意思好像是把0，1号分区互相换个Broker

<table> 
 <thead> 
  <tr> 
   <th>Broker-1</th> 
   <th>Broker-2</th> 
   <th>Broker-3</th> 
   <th>Broker-4</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td>0</td> 
   <td>2</td> 
   <td>3</td> 
  </tr> 
 </tbody> 
</table>

但是实际上不会这样做,Controller在处理的时候会把前面3个截掉; 只取新增的分区分配方式,原来的还是不会变

<table> 
 <thead> 
  <tr> 
   <th>Broker-1</th> 
   <th>Broker-2</th> 
   <th>Broker-3</th> 
   <th>Broker-4</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>0</td> 
   <td>1</td> 
   <td>2</td> 
   <td>3</td> 
  </tr> 
 </tbody> 
</table>

### 1.4.查询Topic描述 ###

**1.查询单个Topic**

> `sh bin/kafka-topics.sh --topic test --bootstrap-server xxxx:9092 --describe --exclude-internal`

**2.批量查询Topic**(正则表达式匹配,下面是查询所有Topic)

> `sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server xxxx:9092 --describe --exclude-internal`

支持正则表达式匹配Topic,只需要将topic 用双引号包裹起来

--------------------

相关可选参数

### 5.查询Topic列表 ###

**1.查询所有Topic列表**

> `sh bin/kafka-topics.sh --bootstrap-server xxxxxx:9092 --list --exclude-internal`

**2.查询匹配Topic列表**(正则表达式)

> 查询`test_create_`开头的所有Topic列表  
> `sh bin/kafka-topics.sh --bootstrap-server xxxxxx:9092 --list --exclude-internal --topic "test_create_.*"`

--------------------

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--exclude-internal</code></td> 
   <td>排除kafka内部topic,比如<code>__consumer_offsets-*</code></td> 
   <td><code>--exclude-internal</code></td> 
  </tr> 
  <tr> 
   <td><code>--topic</code></td> 
   <td>可以正则表达式进行匹配,展示topic名称</td> 
   <td><code>--topic</code></td> 
  </tr> 
 </tbody> 
</table>

## 2.ConfigCommand ##

> Config相关操作; 动态配置可以覆盖默认的静态配置;

### 2.1 查询配置 ###

#### Topic配置查询 ####

> 展示关于Topic的动静态配置

**1.查询单个Topic配置**(只列举动态配置)

> `sh bin/kafka-configs.sh --describe --bootstrap-server xxxxx:9092 --topic test_create_topic`  
> 或者  
> `sh bin/kafka-configs.sh --describe --bootstrap-server 172.23.248.85:9092 --entity-type topics --entity-name test_create_topic`

**2.查询所有Topic配置**(包括内部Topic)(只列举动态配置)

> `sh bin/kafka-configs.sh --describe --bootstrap-server 172.23.248.85:9092 --entity-type topics`

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70]  
![在这里插入图片描述][d10ce9b066f14c8b9a491aadb86ec1c5.png]

**3.查询Topic的详细配置(动态+静态)**

> 只需要加上一个参数`--all`

#### 其他配置/clients/users/brokers/broker-loggers 的查询 ####

> 同理 ；只需要将`--entity-type` 改成对应的类型就行了 (topics/clients/users/brokers/broker-loggers)

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 1]

#### 查询kafka版本信息 ####

> `sh bin/kafka-configs.sh --describe --bootstrap-server xxxx:9092 --version`

**所有可配置的动态配置 请看最后面的 *附件* 部分**

### 2.2 增删改 配置 `--alter` ###

**–alter**

**删除配置**: `--delete-config` k1=v1,k2=v2  
**添加/修改配**置: `--add-config` k1,k2  
**选择类型**: `--entity-type` (topics/clients/users/brokers/broker-  
loggers)  
**类型名称**: `--entity-name`

#### Topic添加/修改动态配置 ####

`--add-config`

> `sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --add-config file.delete.delay.ms=222222,retention.ms=999999`

#### Topic删除动态配置 ####

`--delete-config`

> `sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --delete-config file.delete.delay.ms,retention.ms`

#### 添加/删除配置同时执行 ####

> `sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type brokers --entity-default --add-config log.segment.bytes=788888888 --delete-config log.retention.ms`

#### 其他配置同理,只需要类型改下`--entity-type` ####

> 类型有: (topics/clients/users/brokers/broker- loggers)

哪些配置可以修改 请看最后面的附件：**ConfigCommand 的一些可选配置**

### 默认配置 ###

配置默认 **`--entity-default`**

> `sh bin/kafka-configs.sh --bootstrap-server xxxxx:9090 --alter --entity-type brokers --entity-default --add-config log.segment.bytes=88888888`

动态配置的默认配置是使用了节点 `<defalut>`;

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 2]  
该图转自https://www.cnblogs.com/lizherui/p/12271285.html

**优先级** 指定动态配置>默认动态配置>静态配置

## 3.副本扩缩、分区迁移、跨路径迁移 kafka-reassign-partitions ##

请戳 [【kafka运维】副本扩缩容、数据迁移、副本重分配、副本跨路径迁移][kafka]

[kafka实战】分区重分配可能出现的问题和排查问题思路(生产环境实战,干货!!!非常干!!!建议收藏)][kafka 1]

## 4.Topic的发送kafka-console-producer.sh ##

**4.1 生产无key消息**

## 生产者
    bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties

**4.2 生产有key消息**  
加上属性`--property parse.key=true`

## 生产者
    bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties  --property parse.key=true

默认消息key与消息value间使用“Tab键”进行分隔，所以消息key以及value中切勿使用转义字符(\\t)

--------------------

可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>值类型</th> 
   <th>说明</th> 
   <th>有效值</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>–bootstrap-server</td> 
   <td>String</td> 
   <td>要连接的服务器必需(除非指定–broker-list)</td> 
   <td>如：host1:prot1,host2:prot2</td> 
  </tr> 
  <tr> 
   <td>–topic</td> 
   <td>String</td> 
   <td>(必需)接收消息的主题名称</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>–batch-size</td> 
   <td>Integer</td> 
   <td>单个批处理中发送的消息数</td> 
   <td>200(默认值)</td> 
  </tr> 
  <tr> 
   <td>–compression-codec</td> 
   <td>String</td> 
   <td>压缩编解码器</td> 
   <td>none、gzip(默认值)snappy、lz4、zstd</td> 
  </tr> 
  <tr> 
   <td>–max-block-ms</td> 
   <td>Long</td> 
   <td>在发送请求期间，生产者将阻止的最长时间</td> 
   <td>60000(默认值)</td> 
  </tr> 
  <tr> 
   <td>–max-memory-bytes</td> 
   <td>Long</td> 
   <td>生产者用来缓冲等待发送到服务器的总内存</td> 
   <td>33554432(默认值)</td> 
  </tr> 
  <tr> 
   <td>–max-partition-memory-bytes</td> 
   <td>Long</td> 
   <td>为分区分配的缓冲区大小</td> 
   <td>16384</td> 
  </tr> 
  <tr> 
   <td>–message-send-max-retries</td> 
   <td>Integer</td> 
   <td>最大的重试发送次数</td> 
   <td>3</td> 
  </tr> 
  <tr> 
   <td>–metadata-expiry-ms</td> 
   <td>Long</td> 
   <td>强制更新元数据的时间阈值(ms)</td> 
   <td>300000</td> 
  </tr> 
  <tr> 
   <td>–producer-property</td> 
   <td>String</td> 
   <td>将自定义属性传递给生成器的机制</td> 
   <td>如：key=value</td> 
  </tr> 
  <tr> 
   <td>–producer.config</td> 
   <td>String</td> 
   <td>生产者配置属性文件[–producer-property]优先于此配置 配置文件完整路径</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>–property</td> 
   <td>String</td> 
   <td>自定义消息读取器</td> 
   <td>parse.key=true/false key.separator=&lt;key.separator&gt;ignore.error=true/false</td> 
  </tr> 
  <tr> 
   <td>–request-required-acks</td> 
   <td>String</td> 
   <td>生产者请求的确认方式</td> 
   <td>0、1(默认值)、all</td> 
  </tr> 
  <tr> 
   <td>–request-timeout-ms</td> 
   <td>Integer</td> 
   <td>生产者请求的确认超时时间</td> 
   <td>1500(默认值)</td> 
  </tr> 
  <tr> 
   <td>–retry-backoff-ms</td> 
   <td>Integer</td> 
   <td>生产者重试前，刷新元数据的等待时间阈值</td> 
   <td>100(默认值)</td> 
  </tr> 
  <tr> 
   <td>–socket-buffer-size</td> 
   <td>Integer</td> 
   <td>TCP接收缓冲大小</td> 
   <td>102400(默认值)</td> 
  </tr> 
  <tr> 
   <td>–timeout</td> 
   <td>Integer</td> 
   <td>消息排队异步等待处理的时间阈值</td> 
   <td>1000(默认值)</td> 
  </tr> 
  <tr> 
   <td>–sync</td> 
   <td>同步发送消息</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>–version</td> 
   <td>显示 Kafka 版本</td> 
   <td>不配合其他参数时，显示为本地Kafka版本</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>–help</td> 
   <td>打印帮助信息</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## 5. Topic的消费kafka-console-consumer.sh ##

**1. 新客户端从头消费`--from-beginning` (注意这里是新客户端,如果之前已经消费过了是不会从头消费的)**  
下面没有指定客户端名称,所以每次执行都是新客户端都会从头消费

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

**2. 正则表达式匹配topic进行消费`--whitelist`**  
**`消费所有的topic`**

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --whitelist ‘.\*’

**`消费所有的topic，并且还从头消费`**

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --whitelist ‘.\*’ --from-beginning

**3.显示key进行消费`--property print.key=true`**

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --property print.key=true

**4. 指定分区消费`--partition` 指定起始偏移量消费`--offset`**

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --offset 100

**5. 给客户端命名`--group`**

注意给客户端命名之后,如果之前有过消费，那么`--from-beginning`就不会再从头消费了

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --group test-group

**6. 添加客户端属性`--consumer-property`**

这个参数也可以给客户端添加属性,但是注意 不能多个地方配置同一个属性,他们是互斥的;比如在下面的基础上还加上属性`--group test-group` 那肯定不行

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test `--consumer-property group.id=test-consumer-group`

**7. 添加客户端属性`--consumer.config`**

跟`--consumer-property` 一样的性质,都是添加客户端的属性,不过这里是指定一个文件,把属性写在文件里面, `--consumer-property` 的优先级大于 `--consumer.config`

> sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config config/consumer.properties

--------------------

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--group</code></td> 
   <td>指定消费者所属组的ID</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--topic</code></td> 
   <td>被消费的topic</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--partition</code></td> 
   <td>指定分区 ；除非指定<code>–offset</code>，否则从分区结束(latest)开始消费</td> 
   <td><code>--partition 0</code></td> 
  </tr> 
  <tr> 
   <td><code>--offset</code></td> 
   <td>执行消费的起始offset位置 ;默认值: latest; /latest /earliest /偏移量</td> 
   <td><code>--offset</code> 10</td> 
  </tr> 
  <tr> 
   <td><code>--whitelist</code></td> 
   <td>正则表达式匹配topic；<code>--topic</code>就不用指定了; 匹配到的所有topic都会消费; 当然用了这个参数,<code>--partition</code> <code>--offset</code>等就不能使用了</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--consumer-property</code></td> 
   <td>将用户定义的属性以key=value的形式传递给使用者</td> 
   <td><code>--consumer-property</code>group.id=test-consumer-group</td> 
  </tr> 
  <tr> 
   <td><code>--consumer.config</code></td> 
   <td>消费者配置属性文件请注意，[<code>consumer-property</code>]优先于此配置</td> 
   <td><code>--consumer.config</code> config/consumer.properties</td> 
  </tr> 
  <tr> 
   <td><code>--property</code></td> 
   <td>初始化消息格式化程序的属性</td> 
   <td>print.timestamp=true,false 、print.key=true,false 、print.value=true,false 、key.separator=&lt;key.separator&gt; 、line.separator=&lt;line.separator&gt;、key.deserializer=&lt;key.deserializer&gt;、value.deserializer=&lt;value.deserializer&gt;</td> 
  </tr> 
  <tr> 
   <td><code>--from-beginning</code></td> 
   <td>从存在的最早消息开始，而不是从最新消息开始,注意如果配置了客户端名称并且之前消费过，那就不会从头消费了</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--max-messages</code></td> 
   <td>消费的最大数据量，若不指定，则持续消费下去</td> 
   <td><code>--max-messages</code> 100</td> 
  </tr> 
  <tr> 
   <td><code>--skip-message-on-error</code></td> 
   <td>如果处理消息时出错，请跳过它而不是暂停</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--isolation-level</code></td> 
   <td>设置为read_committed以过滤掉未提交的事务性消息,设置为read_uncommitted以读取所有消息,默认值:read_uncommitted</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--formatter</code></td> 
   <td>kafka.tools.DefaultMessageFormatter、kafka.tools.LoggingMessageFormatter、kafka.tools.NoOpMessageFormatter、kafka.tools.ChecksumMessageFormatter</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## 6.kafka-leader-election Leader重新选举 ##

**6.1 指定Topic指定分区用重新`PREFERRED：优先副本策略` 进行Leader重选举**

> sh bin/kafka-leader-election.sh --bootstrap-server xxxx:9090 --topic test_create_topic4 --election-type PREFERRED --partition 0

**6.2 所有Topic所有分区用重新`PREFERRED：优先副本策略` 进行Leader重选举**

sh bin/kafka-leader-election.sh --bootstrap-server xxxx:9090 --election-type preferred  --all-topic-partitions

**6.3 设置配置文件批量指定topic和分区进行Leader重选举**

先配置leader-election.json文件

{ 
      "partitions": [
        { 
          "topic": "test_create_topic4",
          "partition": 1
        },
        { 
          "topic": "test_create_topic4",
          "partition": 2
        }
      ]
    }

sh bin/kafka-leader-election.sh --bootstrap-server xxx:9090 --election-type preferred  --path-to-json-file config/leader-election.json

--------------------

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--bootstrap-server</code> 指定kafka服务</td> 
   <td>指定连接到的kafka服务</td> 
   <td>–bootstrap-server localhost:9092</td> 
  </tr> 
  <tr> 
   <td><code>--topic</code></td> 
   <td>指定Topic，此参数跟<code>--all-topic-partitions</code>和<code>path-to-json-file</code> 三者互斥</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--partition</code></td> 
   <td>指定分区,跟<code>--topic</code>搭配使用</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--election-type</code></td> 
   <td>两个选举策略(<code>PREFERRED:</code>优先副本选举,如果第一个副本不在线的话会失败;<code>UNCLEAN</code>: 策略)</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--all-topic-partitions</code></td> 
   <td>所有topic所有分区执行Leader重选举; 此参数跟<code>--topic</code>和<code>path-to-json-file</code> 三者互斥</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--path-to-json-file</code></td> 
   <td>配置文件批量选举，此参数跟<code>--topic</code>和<code>all-topic-partitions</code> 三者互斥</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## 7. 持续批量推送消息kafka-verifiable-producer.sh ##

**单次发送100条消息`--max-messages 100`**

一共要推送多少条，默认为-1，-1表示一直推送到进程关闭位置

> sh bin/kafka-verifiable-producer.sh --topic test\_create\_topic4 --bootstrap-server localhost:9092 `--max-messages 100`

**每秒发送最大吞吐量不超过消息 `--throughput 100`**

推送消息时的吞吐量，单位messages/sec。默认为-1，表示没有限制

> sh bin/kafka-verifiable-producer.sh --topic test\_create\_topic4 --bootstrap-server localhost:9092 `--throughput 100`

**发送的消息体带前缀`--value-prefix`**

> sh bin/kafka-verifiable-producer.sh --topic test\_create\_topic4 --bootstrap-server localhost:9092 `--value-prefix 666`

注意`--value-prefix 666`必须是整数,发送的消息体的格式是加上一个 点号`.` 例如： `666.`

其他参数：  
`--producer.config CONFIG_FILE` 指定producer的配置文件  
`--acks ACKS` 每次推送消息的ack值，默认是-1

## 8. 持续批量拉取消息kafka-verifiable-consumer ##

**持续消费**

> sh bin/kafka-verifiable-consumer.sh --group-id test\_consumer --bootstrap-server localhost:9092 --topic test\_create\_topic4

**单次最大消费10条消息`--max-messages 10`**

> sh bin/kafka-verifiable-consumer.sh --group-id test\_consumer --bootstrap-server localhost:9092 --topic test\_create\_topic4 `--max-messages 10`

--------------------

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--bootstrap-server</code> 指定kafka服务</td> 
   <td>指定连接到的kafka服务;</td> 
   <td>–bootstrap-server localhost:9092</td> 
  </tr> 
  <tr> 
   <td><code>--topic</code></td> 
   <td>指定消费的topic</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--group-id</code></td> 
   <td>消费者id；不指定的话每次都是新的组id</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>group-instance-id</code></td> 
   <td>消费组实例ID,唯一值</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--max-messages</code></td> 
   <td>单次最大消费的消息数量</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--enable-autocommit</code></td> 
   <td>是否开启offset自动提交；默认为false</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--reset-policy</code></td> 
   <td>当以前没有消费记录时，选择要拉取offset的策略，可以是<code>earliest</code>, <code>latest</code>,<code>none</code>。默认是earliest</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--assignment-strategy</code></td> 
   <td>consumer分配分区策略，默认是<code>org.apache.kafka.clients.consumer.RangeAssignor</code></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--consumer.config</code></td> 
   <td>指定consumer的配置文件</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## 9.生产者压力测试kafka-producer-perf-test.sh ##

**1. 发送1024条消息`--num-records 100`并且每条消息大小为1KB`--record-size 1024` 最大吞吐量每秒10000条`--throughput 100`**

> sh bin/kafka-producer-perf-test.sh --topic test\_create\_topic4 --num-records 100 --throughput 100000 --producer-props bootstrap.servers=localhost:9092 --record-size 1024

你可以通过[LogIKM][LogI-Kafka-Manager]查看分区是否增加了对应的数据大小  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 3]  
从[LogIKM][LogI-Kafka-Manager] 可以看到发送了1024条消息; 并且总数据量=1M; 1024条\*1024byte = 1M;

**2. 用指定消息文件`--payload-file`发送100条消息最大吞吐量每秒100条`--throughput 100`**

1.  先配置好消息文件`batchmessage.txt`  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 4]
2.  然后执行命令  
    发送的消息会从`batchmessage.txt`里面随机选择; 注意这里我们没有用参数`--payload-delimeter`指定分隔符，默认分隔符是\\n换行;
    
    > bin/kafka-producer-perf-test.sh --topic test\_create\_topic4 --num-records 1024 --throughput 100 --producer-props bootstrap.servers=localhost:9090 --payload-file config/batchmessage.txt
3.  验证消息，可以通过 [LogIKM][LogI-Kafka-Manager] 查看发送的消息
    
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 5]

--------------------

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--topic</code></td> 
   <td>指定消费的topic</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--num-records</code></td> 
   <td>发送多少条消息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--throughput</code></td> 
   <td>每秒消息最大吞吐量</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--producer-props</code></td> 
   <td>生产者配置, k1=v1,k2=v2</td> 
   <td><code>--producer-props</code> bootstrap.servers= localhost:9092,client.id=test_client</td> 
  </tr> 
  <tr> 
   <td><code>--producer.config</code></td> 
   <td>生产者配置文件</td> 
   <td><code>--producer.config</code> config/producer.propeties</td> 
  </tr> 
  <tr> 
   <td><code>--print-metrics</code></td> 
   <td>在test结束的时候打印监控信息,默认false</td> 
   <td><code>--print-metrics</code> true</td> 
  </tr> 
  <tr> 
   <td><code>--transactional-id</code></td> 
   <td>指定事务 ID，测试并发事务的性能时需要，只有在 --transaction-duration-ms &gt; 0 时生效，默认值为 performance-producer-default-transactional-id</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--transaction-duration-ms</code></td> 
   <td>指定事务持续的最长时间，超过这段时间后就会调用 commitTransaction 来提交事务，只有指定了 &gt; 0 的值才会开启事务，默认值为 0</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--record-size</code></td> 
   <td>一条消息的大小byte; 和 --payload-file 两个中必须指定一个，但不能同时指定</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--payload-file</code></td> 
   <td>指定消息的来源文件，只支持 UTF-8 编码的文本文件，文件的消息分隔符通过 <code>--payload-delimeter</code>指定,默认是用换行\nl来分割的，和 --record-size 两个中必须指定一个，但不能同时指定 ; 如果提供的消息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--payload-delimeter</code></td> 
   <td>如果通过 <code>--payload-file</code> 指定了从文件中获取消息内容，那么这个参数的意义是指定文件的消息分隔符，默认值为 \n，即文件的每一行视为一条消息；如果未指定<code>--payload-file</code>则此参数不生效；发送消息的时候是随机送文件里面选择消息发送的;</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## 10.消费者压力测试kafka-consumer-perf-test.sh ##

**消费100条消息`--messages 100`**

> sh bin/kafka-consumer-perf-test.sh -topic test\_create\_topic4 --bootstrap-server localhost:9090 --messages 100

--------------------

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--bootstrap-server</code></td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--consumer.config</code></td> 
   <td>消费者配置文件</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--date-format</code></td> 
   <td>结果打印出来的时间格式化</td> 
   <td>默认：yyyy-MM-dd HH:mm:ss:SSS</td> 
  </tr> 
  <tr> 
   <td><code>--fetch-size</code></td> 
   <td>单次请求获取数据的大小</td> 
   <td>默认1048576</td> 
  </tr> 
  <tr> 
   <td><code>--topic</code></td> 
   <td>指定消费的topic</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--from-latest</code></td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--group</code></td> 
   <td>消费组ID</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--hide-header</code></td> 
   <td>如果设置了,则不打印header信息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--messages</code></td> 
   <td>需要消费的数量</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--num-fetch-threads</code></td> 
   <td>feth 数据的线程数(<code>废弃无效</code>)</td> 
   <td>默认：1</td> 
  </tr> 
  <tr> 
   <td><code>--print-metrics</code></td> 
   <td>结束的时候打印监控数据</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--show-detailed-stats</code></td> 
   <td>如果设置，则按照<code>--report_interval</code>配置的方式报告每个报告间隔的统计信息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--threads</code></td> 
   <td>消费线程数;(<code>废弃无效</code>)</td> 
   <td>默认 10</td> 
  </tr> 
  <tr> 
   <td><code>--reporting-interval</code></td> 
   <td>打印进度信息的时间间隔（以毫秒为单位）</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## 11.删除指定分区的消息kafka-delete-records.sh ##

**删除指定topic的某个分区的消息删除至offset为1024**

先配置json文件`offset-json-file.json`

{ "partitions":
    [{ "topic": "test1", "partition": 0,
      "offset": 1024}],
      "version":1
    }

在执行命令

> sh bin/kafka-delete-records.sh --bootstrap-server 172.23.250.249:9090 --offset-json-file config/offset-json-file.json

验证 通过 [LogIKM][LogI-Kafka-Manager] 查看发送的消息

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 6]  
**从这里可以看出来,配置`"offset": 1024` 的意思是从最开始的地方删除消息到 1024的offset; 是从最前面开始删除的**

## 12. 查看Broker磁盘信息kafka-log-dirs.sh ##

**查询指定topic磁盘信息`--topic-list topic1,topic2`**

> sh bin/kafka-log-dirs.sh --bootstrap-server xxxx:9090 --describe --topic-list test2

**查询指定Broker磁盘信息`--broker-list 0 broker1,broker2`**

> sh bin/kafka-log-dirs.sh --bootstrap-server xxxxx:9090 --describe --topic-list test2 --broker-list 0

例如我一个3分区3副本的Topic的查出来的信息  
`logDir` Broker中配置的`log.dir`

{ 
    	"version": 1,
    	"brokers": [{ 
    		"broker": 0,
    		"logDirs": [{ 
    			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-0",
    			"error": null,
    			"partitions": [{ 
    				"partition": "test2-1",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}, { 
    				"partition": "test2-0",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}, { 
    				"partition": "test2-2",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}]
    		}]
    	}, { 
    		"broker": 1,
    		"logDirs": [{ 
    			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-1",
    			"error": null,
    			"partitions": [{ 
    				"partition": "test2-1",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}, { 
    				"partition": "test2-0",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}, { 
    				"partition": "test2-2",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}]
    		}]
    	}, { 
    		"broker": 2,
    		"logDirs": [{ 
    			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-2",
    			"error": null,
    			"partitions": [{ 
    				"partition": "test2-1",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}, { 
    				"partition": "test2-0",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}, { 
    				"partition": "test2-2",
    				"size": 0,
    				"offsetLag": 0,
    				"isFuture": false
    			}]
    		}]
    	}, { 
    		"broker": 3,
    		"logDirs": [{ 
    			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-3",
    			"error": null,
    			"partitions": []
    		}]
    	}]
    }

如果你觉得通过命令查询磁盘信息比较麻烦，你也可以通过 [LogIKM][LogI-Kafka-Manager] 查看  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 7]

## 12. 消费者组管理 kafka-consumer-groups.sh ##

### 1. 查看消费者列表`--list` ###

> `sh bin/kafka-consumer-groups.sh --bootstrap-server xxxx:9090 --list`  
> ![在这里插入图片描述][20210625175303377.png]

先调用`MetadataRequest`拿到所有在线Broker列表  
再给每个Broker发送`ListGroupsRequest`请求获取 消费者组数据

### 2. 查看消费者组详情`--describe` ###

`DescribeGroupsRequest`

**查看消费组详情`--group` 或 `--all-groups`**

> **查看指定消费组详情`--group`**  
> `sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --group test2_consumer_group`
> 
> --------------------
> 
> **查看所有消费组详情`--all-groups`**  
> `sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --all-groups`  
> 查看该消费组 消费的所有Topic、及所在分区、最新消费offset、Log最新数据offset、Lag还未消费数量、消费者ID等等信息  
> ![在这里插入图片描述][20210625174905816.png]

**查询消费者成员信息`--members`**

> **所有消费组成员信息**  
> `sh bin/kafka-consumer-groups.sh --describe --all-groups --members --bootstrap-server xxx:9090`  
> **指定消费组成员信息**  
> `sh bin/kafka-consumer-groups.sh --describe --members --group test2_consumer_group --bootstrap-server xxxx:9090`  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 8]

**查询消费者状态信息`--state`**

> **所有消费组状态信息**  
> `sh bin/kafka-consumer-groups.sh --describe --all-groups --state --bootstrap-server xxxx:9090`  
> **指定消费组状态信息**  
> `sh bin/kafka-consumer-groups.sh --describe --state --group test2_consumer_group --bootstrap-server xxxxx:9090`  
> ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 9]

### 3. 删除消费者组`--delete` ###

`DeleteGroupsRequest`

**删除消费组–delete**

> **删除指定消费组`--group`**  
> `sh bin/kafka-consumer-groups.sh --delete --group test2_consumer_group --bootstrap-server xxxx:9090`  
> **删除所有消费组`--all-groups`**  
> `sh bin/kafka-consumer-groups.sh --delete --all-groups --bootstrap-server xxxx:9090`

**PS: 想要删除消费组前提是这个消费组的所有客户端都停止消费/不在线才能够成功删除;否则会报下面异常**

Error: Deletion of some consumer groups failed:
    * Group 'test2_consumer_group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.

### 4. 重置消费组的偏移量 `--reset-offsets` ###

能够执行成功的一个前提是 消费组这会是不可用状态;

下面的示例使用的参数是: `--dry-run` ;这个参数表示预执行,会打印出来将要处理的结果;  
等你想真正执行的时候请换成参数`--excute` ;

下面示例 重置模式都是 `--to-earliest` 重置到最早的;

请根据需要参考下面 **相关重置Offset的模式** 换成其他模式;

**重置指定消费组的偏移量 `--group`**

> **重置指定消费组的所有Topic的偏移量`--all-topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --all-topic`  
> **重置指定消费组的指定Topic的偏移量`--topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --topic test2`

**重置所有消费组的偏移量 `--all-group`**

> **重置所有消费组的所有Topic的偏移量`--all-topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --all-topic`  
> **重置所有消费组中指定Topic的偏移量`--topic`**  
> `sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --topic test2`

`--reset-offsets` 后面需要接**重置的模式**

**相关重置Offset的模式**

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><strong><code>--to-earliest</code> :</strong></td> 
   <td>重置offset到最开始的那条offset(找到还未被删除最早的那个offset)</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><strong><code>--to-current</code>:</strong></td> 
   <td>直接重置offset到当前的offset，也就是LOE</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><strong><code>--to-latest</code>：</strong></td> 
   <td>重置到最后一个offset</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><strong><code>--to-datetime</code>:</strong></td> 
   <td>重置到指定时间的offset;格式为:<code>YYYY-MM-DDTHH:mm:SS.sss</code>;</td> 
   <td><code>--to-datetime "2021-6-26T00:00:00.000"</code></td> 
  </tr> 
  <tr> 
   <td><code>--to-offset</code></td> 
   <td>重置到指定的offset,但是通常情况下,匹配到多个分区,这里是将匹配到的所有分区都重置到这一个值; 如果 1.目标最大offset&lt;<code>--to-offset</code>, 这个时候重置为目标最大offset；2.目标最小offset&gt;<code>--to-offset</code> ，则重置为最小; 3.否则的话才会重置为<code>--to-offset</code>的目标值; <strong>一般不用这个</strong></td> 
   <td><code>--to-offset 3465</code> <img src="https://img-blog.csdnimg.cn/20210626153528929.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY=,size_16,color_FFFFFF,t_70" alt="在这里插入图片描述"></td> 
  </tr> 
  <tr> 
   <td><code>--shift-by</code></td> 
   <td>按照偏移量增加或者减少多少个offset；正的为往前增加;负的往后退；当然这里也是匹配所有的;</td> 
   <td><code>--shift-by 100</code> 、<code>--shift-by -100</code></td> 
  </tr> 
  <tr> 
   <td><code>--from-file</code></td> 
   <td>根据CVS文档来重置; 这里下面单独讲解</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

**`--from-file`着重讲解一下**

> 上面其他的一些模式重置的都是匹配到的所有分区; 不能够每个分区重置到不同的offset；不过\*\*`--from-file`\*\*可以让我们更灵活一点;

1.  先配置cvs文档  
    格式为: Topic:分区号: 重置目标偏移量
    
        test2,0,100
        test2,1,200
        test2,2,300
2.  执行命令
    
    > `sh bin/kafka-consumer-groups.sh --reset-offsets --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --from-file config/reset-offset.csv`

### 5. 删除偏移量`delete-offsets` ###

能够执行成功的一个前提是 消费组这会是不可用状态;

偏移量被删除了之后,Consumer Group下次启动的时候,会从头消费;

> `sh bin/kafka-consumer-groups.sh --delete-offsets --group test2_consumer_group2 --bootstrap-server XXXX:9090 --topic test2`

--------------------

相关可选参数

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--bootstrap-server</code></td> 
   <td>指定连接到的kafka服务;</td> 
   <td>–bootstrap-server localhost:9092</td> 
  </tr> 
  <tr> 
   <td><code>--list</code></td> 
   <td>列出所有消费组名称</td> 
   <td><code>--list</code></td> 
  </tr> 
  <tr> 
   <td><code>--describe</code></td> 
   <td>查询消费者描述信息</td> 
   <td><code>--describe</code></td> 
  </tr> 
  <tr> 
   <td><code>--group</code></td> 
   <td>指定消费组</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--all-groups</code></td> 
   <td>指定所有消费组</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--members</code></td> 
   <td>查询消费组的成员信息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--state</code></td> 
   <td>查询消费者的状态信息</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--offsets</code></td> 
   <td>在查询消费组描述信息的时候,这个参数会列出消息的偏移量信息; 默认就会有这个参数的;</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>dry-run</code></td> 
   <td>重置偏移量的时候,使用这个参数可以让你预先看到重置情况，这个时候还没有真正的执行,真正执行换成<code>--excute</code>;默认为<code>dry-run</code></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--excute</code></td> 
   <td>真正的执行重置偏移量的操作;</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--to-earliest</code></td> 
   <td>将offset重置到最早</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>to-latest</code></td> 
   <td>将offset重置到最近</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

## 13.查看日志文件 kafka-dump-log.sh ##

<table> 
 <thead> 
  <tr> 
   <th>参数</th> 
   <th>描述</th> 
   <th>例子</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td><code>--deep-iteration</code></td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--files &lt;String: file1, file2, ...&gt;</code></td> 
   <td>必需; 读取的日志文件</td> 
   <td>–files 0000009000.log</td> 
  </tr> 
  <tr> 
   <td><code>--key-decoder-class</code></td> 
   <td>如果设置，则用于反序列化键。这类应实现kafka.serializer。解码器特性。自定义jar应该是在kafka/libs目录中提供</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--max-message-size</code></td> 
   <td>最大的数据量,默认：5242880</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--offsets-decoder</code></td> 
   <td>if set, log data will be parsed as offset data from the __consumer_offsets topic.</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--print-data-log</code></td> 
   <td>打印内容</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--transaction-log-decoder</code></td> 
   <td>if set, log data will be parsed as transaction metadata from the __transaction_state topic</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--value-decoder-class [String]</code></td> 
   <td>if set, used to deserialize the messages. This class should implement kafka. serializer.Decoder trait. Custom jar should be available in kafka/libs directory. (default: kafka.serializer. StringDecoder)</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td><code>--verify-index-only</code></td> 
   <td>if set, just verify the index log without printing its content.</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

**查询Log文件**

> `sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.log`  
> ![在这里插入图片描述][e90ad7f1c39e5208120080475845dbb3.png]

**查询Log文件具体信息 `--print-data-log`**

> `sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.log --print-data-log`  
> ![在这里插入图片描述][3306f80db58a943e6a59c39e66b4c4fe.png]

**查询index文件具体信息**

> `sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.index`  
> ![在这里插入图片描述][bd918af516229bbb80640483831d6595.png]  
> 配置项为`log.index.size.max.bytes`； 来控制创建索引的大小;

**查询timeindex文件**

> `sh bin/kafka-dump-log.sh --files kafka-logs-0/test2-0/00000000000000000300.timeindex`  
> ![在这里插入图片描述][2254d080ca4e63d13714b40b6c0c3b63.png]

## 附件 ##

**ConfigCommand 的一些可选配置**

--------------------

Topic相关可选配置

<table> 
 <thead> 
  <tr> 
   <th>key</th> 
   <th>value</th> 
   <th>示例</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>cleanup.policy</td> 
   <td>清理策略</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td>压缩类型(通常建议在produce端控制)</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>delete.retention.ms</td> 
   <td>压缩日志的保留时间</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>file.delete.delay.ms</td> 
   <td>topic删除被标记为–delete文件之后延迟多长时间删除正在的Log文件</td> 
   <td>60000</td> 
  </tr> 
  <tr> 
   <td>flush.messages</td> 
   <td>持久化message限制</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>flush.ms</td> 
   <td>持久化频率</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>follower.replication.throttled.replicas</td> 
   <td>flowwer副本限流 格式：分区号:副本follower号,分区号:副本follower号</td> 
   <td>0:1,1:1</td> 
  </tr> 
  <tr> 
   <td>index.interval.bytes</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>leader.replication.throttled.replicas</td> 
   <td>leader副本限流 格式：分区号:副本Leader号</td> 
   <td>0:0</td> 
  </tr> 
  <tr> 
   <td>max.compaction.lag.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>max.message.bytes</td> 
   <td>最大的batch的message大小</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>message.downconversion.enable</td> 
   <td>message是否向下兼容</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>message.format.version</td> 
   <td>message格式版本</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>message.timestamp.difference.max.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>message.timestamp.type</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>min.cleanable.dirty.ratio</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>min.compaction.lag.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>min.insync.replicas</td> 
   <td>最小的ISR</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>preallocate</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>retention.bytes</td> 
   <td>日志保留大小(通常按照时间限制)</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>retention.ms</td> 
   <td>日志保留时间</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>segment.bytes</td> 
   <td>segment的大小限制</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>segment.index.bytes</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>segment.jitter.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>segment.ms</td> 
   <td>segment的切割时间</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>unclean.leader.election.enable</td> 
   <td>是否允许非同步副本选主</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

Broker相关可选配置

<table> 
 <thead> 
  <tr> 
   <th>key</th> 
   <th>value</th> 
   <th>示例</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>advertised.listeners</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>background.threads</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>compression.type</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>follower.replication.throttled.rate</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>leader.replication.throttled.rate</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>listener.security.protocol.map</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>listeners</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.backoff.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.dedupe.buffer.size</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.delete.retention.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.io.buffer.load.factor</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.io.buffer.size</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.io.max.bytes.per.second</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.max.compaction.lag.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.min.cleanable.ratio</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.min.compaction.lag.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleaner.threads</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.cleanup.policy</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.flush.interval.messages</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.flush.interval.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.index.interval.bytes</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.index.size.max.bytes</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.message.downconversion.enable</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.message.timestamp.difference.max.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.message.timestamp.type</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.preallocate</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.retention.bytes</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.retention.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.roll.jitter.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.roll.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.segment.bytes</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>log.segment.delete.delay.ms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>max.connections</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>max.connections.per.ip</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>max.connections.per.ip.overrides</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>message.max.bytes</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>metric.reporters</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>min.insync.replicas</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>num.io.threads</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>num.network.threads</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>num.recovery.threads.per.data.dir</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>num.replica.fetchers</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>principal.builder.class</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>replica.alter.log.dirs.io.max.bytes.per.second</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.enabled.mechanisms</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.jaas.config</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.kinit.cmd</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.min.time.before.relogin</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.principal.to.local.rules</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.service.name</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.ticket.renew.jitter</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.kerberos.ticket.renew.window.factor</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.login.refresh.buffer.seconds</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.login.refresh.min.period.seconds</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.login.refresh.window.factor</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.login.refresh.window.jitter</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>sasl.mechanism.inter.broker.protocol</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.cipher.suites</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.client.auth</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.enabled.protocols</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.endpoint.identification.algorithm</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.key.password</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keymanager.algorithm</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.location</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.password</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.keystore.type</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.protocol</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.provider</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.secure.random.implementation</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.trustmanager.algorithm</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.location</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.password</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>ssl.truststore.type</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>unclean.leader.election.enable</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

Users相关可选配置

<table> 
 <thead> 
  <tr> 
   <th>key</th> 
   <th>value</th> 
   <th>示例</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>SCRAM-SHA-256</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>SCRAM-SHA-512</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>consumer_byte_rate</td> 
   <td>针对消费者user进行限流</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>producer_byte_rate</td> 
   <td>针对生产者进行限流</td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>request_percentage</td> 
   <td>请求百分比</td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

clients相关可选配置

<table> 
 <thead> 
  <tr> 
   <th>key</th> 
   <th>value</th> 
   <th>示例</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>consumer_byte_rate</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>producer_byte_rate</td> 
   <td></td> 
   <td></td> 
  </tr> 
  <tr> 
   <td>request_percentage</td> 
   <td></td> 
   <td></td> 
  </tr> 
 </tbody> 
</table>

**以上大部分运维操作,都可以使用 [LogI-Kafka-Manager][] 在平台上可视化操作;**

--------------------

欢迎 Star和 共建由 滴滴开源的kafka的管理平台,非常优秀非常好用的一款kafka管理平台  
满足所有开发运维日常需求

[滴滴开源Logi-KafkaManager 一站式Kafka监控与管控平台][LogI-Kafka-Manager]

## More ##

Kafka专栏持续更新中…(源码、原理、实战、运维、视频、面试视频)

--------------------

[【kafka运维】Kafka全网最全最详细运维命令合集(精品强烈建议收藏!!!)\_石臻臻的杂货铺-CSDN博客][kafka_Kafka_-CSDN]

[【kafka实战】分区重分配可能出现的问题和排查问题思路(生产环境实战,干货!!!非常干!!!建议收藏)][kafka 2]

[【kafka异常】kafka 常见异常处理方案(持续更新! 建议收藏)][kafka_kafka _]

[【kafka运维】分区从分配、数据迁移、副本扩缩容 (附教学视频)][kafka_]

[【kafka源码】ReassignPartitionsCommand源码分析(副本扩缩、数据迁移、副本重分配、副本跨路径迁移][kafka_ReassignPartitionsCommand]

[【kafka】点击更多…][kafka 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/f6fb20bc849b48dab84ee127f6883e40.png
[Kafka]: https://docs.qq.com/mind/DVWxlaFVGUmhEeFNI
[LogiKM_Kafka]: https://github.com/didi/LogiKM
[LogI-Kafka-Manager]: https://github.com/didi/Logi-KafkaManager
[Link 1]: https://www.runoob.com/regexp/regexp-syntax.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70]: /images/20221005/9414b9e1848d4f0a823db68456662d50.png
[d10ce9b066f14c8b9a491aadb86ec1c5.png]: /images/20221005/abd1b61741ef49ce9ce98f95de0bfade.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 1]: /images/20221005/78220666cbc94fd7aefc9eb71463ce92.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 2]: /images/20221005/9ec3bf9fe1724826b9ab5f2676b5219b.png
[kafka]: https://blog.csdn.net/u010634066/article/details/118028403
[kafka 1]: https://blog.csdn.net/u010634066/article/details/118631272
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 3]: /images/20221005/763987f351694eb382088fac8d1f0661.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 4]: /images/20221005/972aa147920247a69746a579fec9604a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 5]: /images/20221005/01644dc3bc214edcb19096a1a2d68aeb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 6]: /images/20221005/70f5a2478344422a88010645dc35b2f7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 7]: /images/20221005/92bf81839f844256b97dc11bd320829f.png
[20210625175303377.png]: /images/20221005/7c44feca3c404a83a8f4123cf34b6b74.png
[20210625174905816.png]: /images/20221005/74d9bc953026415b867f778156cbf89c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 8]: /images/20221005/9b1a3d79a0884c57874beb630de6b0b2.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2MzQwNjY_size_16_color_FFFFFF_t_70 9]: /images/20221005/5fe003deaf804c14bfe01766a1a897ce.png
[e90ad7f1c39e5208120080475845dbb3.png]: /images/20221005/8b7b4cb74fa14173996480dd51a776df.png
[3306f80db58a943e6a59c39e66b4c4fe.png]: /images/20221005/d5546716e24346b08132c2ed8dc27238.png
[bd918af516229bbb80640483831d6595.png]: /images/20221005/c2f646641ca74e8abc9f7c63ce8ffbaf.png
[2254d080ca4e63d13714b40b6c0c3b63.png]: /images/20221005/2922263ed60c4ba6b80ef3364d23b11b.png
[kafka_Kafka_-CSDN]: https://blog.csdn.net/u010634066/article/details/118215928?spm=1001.2014.3001.5501
[kafka 2]: https://blog.csdn.net/u010634066/article/details/118631272?spm=1001.2014.3001.5501
[kafka_kafka _]: https://blog.csdn.net/u010634066/article/details/118105676?spm=1001.2014.3001.5501
[kafka_]: https://blog.csdn.net/u010634066/article/details/118028403?spm=1001.2014.3001.5501
[kafka_ReassignPartitionsCommand]: https://blog.csdn.net/u010634066/article/details/118051963
[kafka 3]: https://mp.weixin.qq.com/mp/appmsgalbum?__biz=Mzg4ODY1NTcxNg==&action=getalbum&album_id=1966026980307304450#wechat_redirect