Kafka 常用工具脚本总结

系统

启动 Kafka

​​-daemon​​ 参数可以让 Kafka 在后台运行。

kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

复制

指定 JMX 端口启动

JMX 的全称为 Java Management Extensions。顾名思义，是管理 Java 的一种扩展，通过 JMX 可以方便我们监控 Kafka 的内存，线程，CPU 的使用情况，以及生产和消费消息的指标。

JMX_PORT=9999 kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties

复制

停止 Kafka

kafka-server-stop.sh

复制

Topic

创建 Topic

kafka-topics.sh --create  --bootstrap-server 10.37.62.20:9092 --replication-factor 3 --partitions 3 --topic <topic-name>

复制

列出所有 Topic

kafka-topics.sh  --bootstrap-server 10.37.62.20:9092 --list

复制

查看指定 Topic

kafka-topics.sh --bootstrap-server 10.37.62.20:9092 --describe --topic <topic-name>

复制

删除指定 Topic

kafka-topics.sh --bootstrap-server 10.37.62.20:9092 --delete --topic <topic-name>

复制

扩展 Topic 的 Partition 数量

artition 数量只能扩大不能缩小。

kafka-topics.sh --bootstrap-server 10.37.62.20:9092 --topic app --alter --partitions 30

复制

扩展 topic 每个 partition 的副本数量

replication factor 可以扩大也可以缩小，最多不能超过 broker 数量。先创建一个文件名为 increace-factor.json，这里要扩展的是 mysql-audit-log 这个 topic 的 partition 到 15 个：0，1，2 为 broker id。

{"version":1,
"partitions":[
{"topic":"mysql-audit-log","partition":0,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":1,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":2,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":3,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":4,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":5,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":6,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":7,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":8,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":9,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":10,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":11,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":12,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":13,"replicas":[0,1,2]},
{"topic":"mysql-audit-log","partition":14,"replicas":[0,1,2]}
]}

复制

kafka-reassign-partitions.sh --zookeeper 10.37.62.20:2181 --reassignment-json-file  increace-factor.json --execute

复制

查看 Topic 数据大小

#方法一
kafka-log-dirs.sh \
  --bootstrap-server 192.168.1.87:9092 \
  --topic-list mytopic \
  --describe \
  | grep -oP '(?<=size":)\d+'  \
  | awk '{ sum += $1 } END { print sum }'
  
#返回结果，单位 Byte
648

#方法二，需要安装 jq
kafka-log-dirs.sh \
    --bootstrap-server 192.168.1.87:9092 \
    --topic-list mytopic \
    --describe \
  | grep '^{' \
  | jq '[ ..|.size? | numbers ] | add'

#返回结果，单位 Byte
648

复制

消费者组 Consumer Group

列出所有的 Consumer Group

kafka-consumer-groups.sh --bootstrap-server 10.37.62.20:9092 --list

复制

查看指定 Consumer Group 详情

• GROUP:消费者 group
• TOPIC:话题 id
• PARTITION:分区 id
• CURRENT-OFFSET:当前已消费的条数
• LOG-END-OFFSET:总条数
• LAG:未消费的条数
• CONSUMER-ID:消费者 id
• HOST:消费者 ip 地址
• CLIENT-ID:客户端 id

#这里查看的是 logstash_mysql 这个消费者 group 的消费情况
kafka-consumer-groups.sh --bootstrap-server 10.37.62.20:9092 --describe --group logstash_mysql

#返回结果
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                      HOST            CLIENT-ID
logstash_mysql  mysql-audit-log 11         1312115         1312857         742             logstash-5-0545a8a7-f7bd-430c-b619-7a2b206addd2  /10.37.62.24    logstash-5
logstash_mysql  mysql-audit-log 1          1312593         1313345         752             logstash-0-d86bd51a-d010-45de-aa6f-f6da8542b779  /10.37.62.23    logstash-0
logstash_mysql  mysql-audit-log 2          1309548         1310317         769             logstash-1-496340ea-935d-444d-a184-51d42e225054  /10.37.62.24    logstash-1
logstash_mysql  mysql-audit-log 12         1313083         1313194         111             logstash-6-806b20cb-a9af-49c1-b37d-ccb33a646ab2  /10.37.62.24    logstash-6
logstash_mysql  mysql-audit-log 6          1310984         1311192         208             logstash-13-8d474bf6-e8d0-4b8a-b319-cf5e2e6cc078 /10.37.62.24    logstash-13
logstash_mysql  mysql-audit-log 9          1312998         1313768         770             logstash-3-29863fb0-6708-4fb1-9e28-bd81c30ce8ef  /10.37.62.24    logstash-3
logstash_mysql  mysql-audit-log 4          1315150         1315276         126             logstash-11-6d66a188-85b7-476b-bd89-5423ef48cd01 /10.37.62.24    logstash-11
logstash_mysql  mysql-audit-log 0          22770935522     22770935650     128             logstash-0-7be475d6-a49e-4ff9-bf83-6b83f6067306  /10.37.62.24    logstash-0
logstash_mysql  mysql-audit-log 8          1309956         1310103         147             logstash-2-3c313c6f-8c98-4333-8bad-2f9696457d7d  /10.37.62.24    logstash-2
logstash_mysql  mysql-audit-log 13         1314659         1314775         116             logstash-7-e98fd14e-e7f6-45e5-8ccf-2442058f0bc9  /10.37.62.24    logstash-7
logstash_mysql  mysql-audit-log 14         1313145         1313250         105             logstash-8-2c3345a8-f8f1-4f08-a18e-333dff2f0d65  /10.37.62.24    logstash-8
logstash_mysql  mysql-audit-log 5          1314037         1314297         260             logstash-12-ce018227-9e59-4137-a23f-5ccc0c7d4f6a /10.37.62.24    logstash-12
logstash_mysql  mysql-audit-log 10         1312883         1312962         79              logstash-4-9eb84ae4-3351-4083-9b1f-288910a6c3b8  /10.37.62.24    logstash-4
logstash_mysql  mysql-audit-log 7          1312476         1313200         724             logstash-14-680c982e-5cf3-406b-810a-4d5c96b5bdee /10.37.62.24    logstash-14
logstash_mysql  mysql-audit-log 3          1313227         1313328         101             logstash-10-e212dc18-a2bb-42d9-9d0b-095a93841efc /10.37.62.24    logstash-10

复制

删除指定 Consumer Group

kafka-topics.sh --bootstrap-server 10.37.62.20:9092 --delete --topic pgw-nginx

复制

消息

生产消息

普通生产消息

kafka-console-producer.sh --broker-list 11.8.36.125:9092 --topic mytopic
>this is my topic

复制

生产消息指定 Key

​​key.separator=,​​ 指定以逗号作为 key 和 value 的分隔符。

kafka-console-producer.sh --broker-list kafka1:9092 --topic cr7-topic --property parse.key=true --property key.separator=,

>mykey,{"orderAmount":1000,"orderId":1,"productId":101,"productNum":1}

复制

消费消息

从头开始消费

从头开始消费是可以消费到之前的消息的，通过 ​​--from-beginning​​ 指定：

kafka-console-consumer.sh --bootstrap-server 11.8.36.125:9092 --topic mytopic --from-beginning
this is my topic

复制

从尾部开始消费

​​--offset latest​​ 指定从尾部开始消费，另外还需要指定 partition，可以指定多个：

kafka-console-consumer.sh --bootstrap-server 11.8.36.125:9092 --topic mytopic  --offset latest  --partition 0 1 2

复制

消费指定条数的消息

​​--max-messages​​指定取的个数：

kafka-console-consumer.sh --bootstrap-server 11.8.36.125:9092 --topic mytopic  --offset latest  --partition 0 1 2 --max-messages 2
bobo
1111
Processed a total of 2 messages

复制

指定消费组进行消费

​​--consumer-property group.id=<消费者组名>​​执行消费者组进行消费：

kafka-console-consumer.sh --bootstrap-server  kafka1:9092 --topic test_partition --consumer-property  group.id=test_group --from-beginning

复制

查看消息具体内容

kafka-dump-log.sh --files cr7-topic-0/00000000000000000000.log  -deep-iteration --print-data-log
 
 #输出结果
| offset: 1080 CreateTime: 1615020877664 keysize: 1 valuesize: 63 sequence: -1 headerKeys: [] key: 1 payload: {"orderAmount":1000,"orderId":1,"productId":101,"productNum":1}
| offset: 1081 CreateTime: 1615020877677 keysize: 1 valuesize: 63 sequence: -1 headerKeys: [] key: 5 payload: {"orderAmount":1000,"orderId":5,"productId":105,"productNum":5}
| offset: 1082 CreateTime: 1615020877677 keysize: 1 valuesize: 63 sequence: -1 headerKeys: [] key: 7 payload: {"orderAmount":1000,"orderId":7,"productId":107,"productNum":7}
| offset: 1083 CreateTime: 1615020877677 keysize: 1 valuesize: 63 sequence: -1 headerKeys: [] key: 8 payload: {"orderAmount":1000,"orderId":8,"productId":108,"productNum":8}
| offset: 1084 CreateTime: 1615020877677 keysize: 2 valuesize: 65 sequence: -1 headerKeys: [] key: 11 payload: {"orderAmount":1000,"orderId":11,"productId":111,"productNum":11}
| offset: 1085 CreateTime: 1615020877677 keysize: 2 valuesize: 65 sequence: -1 headerKeys: [] key: 15 payload: {"orderAmount":1000,"orderId":15,"productId":115,"productNum":15}
| offset: 1086 CreateTime: 1615020877678 keysize: 2 valuesize: 65 sequence: -1 headerKeys: [] key: 17 payload: {"orderAmount":1000,"orderId":17,"productId":117,"productNum":17}
| offset: 1087 CreateTime: 1615020877678 keysize: 2 valuesize: 65 sequence: -1 headerKeys: [] key: 21 payload: {"orderAmount":1000,"orderId":21,"productId":121,"productNum":21}

复制

查看 Topic 中当前消息总数

Kafka 自带的命令没有直接提供这样的功能，要使用 Kafka 提供的工具类 GetOffsetShell 来计算给定 Topic 每个分区当前最早位移和最新位移，差值就是每个分区的当前的消息总数，将该 Topic 所有分区的消息总数累加就能得到该 Topic 总的消息数。

首先查询 Topic 中每个分区 offset 的最小值（起始位置），使用 ​​--time -2​​ 参数。一个分区的起始位置并不是每时每刻都为 0 ，因为日志清理的动作会清理旧的数据，所以分区的起始位置会自然而然地增加。

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list kafka1:9092 -topic test-topic  --time -2

#前面是分区号，后面是 offset
test-topic:0:0
test-topic:1:0

复制

然后使用​​--time -1​​ 参数查询 Topic 各个分区的 offset 的最大值。

kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list kafka1:9092 --time -1 --topic test-topic

#输出结果
test-topic:0:5500000
test-topic:1:5500000

复制

对于本例来说，test-topic 中当前总的消息数为 (5500000 - 0) + （5500000 - 0），等于 1100 万条。如果只是要获取 Topic 中总的消息数（包括已经从 Kafka 删除的消息），那么只需要将 Topic 中每个 Partition 的 Offset 累加即可。

Offset

重置消费者 Offset

#查看消费者组消费情况
#目前的 0 分区 CURRENT-OFFSET 是 4，2 分区 CURRENT-OFFSET 是 6
kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group my-consumer-group

#返回结果
Consumer group 'my-consumer-group' has no active members.

GROUP             TOPIC                 PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
my-consumer-group transaction-topic-msg 2          6               6               0               -               -               -
my-consumer-group transaction-topic-msg 1          0               0               0               -               -               -
my-consumer-group transaction-topic-msg 0          4               4               0               -               -               -         -

#重置消费者组 offset 为 3，重置是所有分区一起重置
kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group my-consumer-group --reset-offsets --execute --to-offset 3 --topic transaction-topic-msg

#返回结果
[2021-06-25 10:44:51,848] WARN New offset (3) is higher than latest offset for topic partition transaction-topic-msg-1. Value will be set to 0 (kafka.admin.ConsumerGroupCommand$)

GROUP                          TOPIC                          PARTITION  NEW-OFFSET     
my-consumer-group              transaction-topic-msg          0          3              
my-consumer-group              transaction-topic-msg          1          0              
my-consumer-group              transaction-topic-msg          2          3              

#可以看到 0 分区和 2 分区的 CURRENT-OFFSET 都变为 3 了
kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group my-consumer-group

#返回结果
Consumer group 'my-consumer-group' has no active members.

GROUP             TOPIC                 PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID
my-consumer-group transaction-topic-msg 2          3               6               3               -               -               -
my-consumer-group transaction-topic-msg 1          0               0               0               -               -               -
my-consumer-group transaction-topic-msg 0          3               4               1               -               -               -

#可以重新消费到之前的数据
kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic transaction-topic-msg  --group my-consumer-group 

#返回结果
message-111111
message-333333

复制

性能测试

生产者性能测试

• --num-records 10000000: 向指定主题发送了 1 千万条消息。
• --record-size 1024: 每条消息的大小为 1024KB。
• --throughput -1: 不限制吞吐量。
• --producer-props: 指定生产者参数。

• acks=-1: 这要求 ISR 列表里跟 leader 保持同步的那些 follower 都要把消息同步过去，才能认为这条消息是写入成功。
• linger.ms=2000: batch.size 和 linger.ms 是对 kafka producer 性能影响比较大的两个参数。batch.size 是 producer 批量发送的基本单位，默认是 16384Bytes，即 16kB；lingger.ms 是 sender 线程在检查 batch 是否 ready 时候，判断有没有过期的参数，默认大小是 0ms。
• compression.type=lz4: 使用 lz4 压缩算法。

[root@kafka1 ~]# kafka-producer-perf-test.sh --topic test_producer_perf --num-records 10000000 --throughput -1 --record-size 1024 --producer-props bootstrap.servers=kafka1:9092 acks=-1 linger.ms=2000 compression.type=lz4

#输出结果
705600 records sent, 141063.6 records/sec (137.76 MB/sec), 54.8 ms avg latency, 557.0 ms max latency.
1204178 records sent, 240739.3 records/sec (235.10 MB/sec), 44.1 ms avg latency, 402.0 ms max latency.
1370938 records sent, 274187.6 records/sec (267.76 MB/sec), 27.9 ms avg latency, 311.0 ms max latency.
1464605 records sent, 292628.4 records/sec (285.77 MB/sec), 19.2 ms avg latency, 139.0 ms max latency.
1477239 records sent, 295447.8 records/sec (288.52 MB/sec), 31.8 ms avg latency, 290.0 ms max latency.
1446682 records sent, 289336.4 records/sec (282.56 MB/sec), 26.4 ms avg latency, 281.0 ms max latency.
1555098 records sent, 311019.6 records/sec (303.73 MB/sec), 37.6 ms avg latency, 344.0 ms max latency.
10000000 records sent, 263894.020162 records/sec (257.71 MB/sec), 32.60 ms avg latency, 557.00 ms max latency, 12 ms 50th, 140 ms 95th, 262 ms 99th, 396 ms 99.9th.

复制

我们应该关心延时的概率分布情况，仅仅知道一个平均值是没有意义的。这就是这里计算分位数的原因。通常我们关注到 99th 分位就可以了。比如在上面的输出中，99th 值是 262 ms，这表明测试生产者生产的消息中，有 99% 消息的延时都在 262 ms 以内。你完全可以把这个数据当作这个生产者对外承诺的 SLA。

消费者性能测试

[root@kafka1 ~]# kafka-consumer-perf-test.sh --broker-list kafka1:9092 --messages 10000000 --topic test_producer_perf

#输出结果
start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2021-03-09 10:34:18:447, 2021-03-09 10:34:33:948, 9765.6250, 629.9997, 10000000, 645119.6697, 1615257259068, -1615257243567, -0.0000, -0.0062

复制

虽然输出格式有所差别，但该脚本也会打印出消费者的吞吐量数据。比如本例中的 629.9997MB/s。有点令人遗憾的是，它没有计算不同分位数下的分布情况。因此，在实际使用过程中，这个脚本的使用率要比生产者性能测试脚本的使用率低。

修改动态参数

查看支持的动态参数

如果你想要知道动态 Broker 参数都有哪些，一种方式是在 Kafka 官网中查看 Broker 端参数列表，另一种方式是直接运行无参数的 kafka-configs 脚本，该脚本的说明文档会告诉你当前动态 Broker 参数都有哪些。

[root@kafka1 ~]# kafka-configs.sh 
This tool helps to manipulate and describe entity config for a topic, client, user or broker
Option                                 Description                            
------                                 -----------                            
--add-config <String>                  Key Value pairs of configs to add.     
                                         Square brackets can be used to group 
                                         values which contain commas: 'k1=v1, 
                                         k2=[v1,v2,v2],k3=v3'. The following  
                                         #Topic 动态参数
                                         is a list of valid configurations:   
                                         For entity-type 'topics':            
                                        cleanup.policy                        
                                        compression.type                      
                                        delete.retention.ms                   
                                        file.delete.delay.ms                  
                                        flush.messages                        
                                        flush.ms                              
                                        follower.replication.throttled.       
                                         replicas                             
                                        index.interval.bytes                  
                                        leader.replication.throttled.replicas 
                                        max.compaction.lag.ms                 
                                        max.message.bytes                     
                                        message.downconversion.enable         
                                        message.format.version                
                                        message.timestamp.difference.max.ms   
                                        message.timestamp.type                
                                        min.cleanable.dirty.ratio             
                                        min.compaction.lag.ms                 
                                        min.insync.replicas                   
                                        preallocate                           
                                        retention.bytes                       
                                        retention.ms                          
                                        segment.bytes                         
                                        segment.index.bytes                   
                                        segment.jitter.ms                     
                                        segment.ms                            
                                        unclean.leader.election.enable        
                                        #Broker 动态参数
                                       For entity-type 'brokers':             
                                        advertised.listeners                  
                                        background.threads                    
                                        compression.type                      
                                        follower.replication.throttled.rate   
                                        leader.replication.throttled.rate     
                                        listener.security.protocol.map        
                                        listeners                             
                                        log.cleaner.backoff.ms                
                                        log.cleaner.dedupe.buffer.size        
                                        log.cleaner.delete.retention.ms       
                                        log.cleaner.io.buffer.load.factor     
                                        log.cleaner.io.buffer.size            
                                        log.cleaner.io.max.bytes.per.second   
                                        log.cleaner.max.compaction.lag.ms     
                                        log.cleaner.min.cleanable.ratio       
                                        log.cleaner.min.compaction.lag.ms     
                                        log.cleaner.threads                   
                                        log.cleanup.policy                    
                                        log.flush.interval.messages           
                                        log.flush.interval.ms                 
                                        log.index.interval.bytes              
                                        log.index.size.max.bytes              
                                        log.message.downconversion.enable     
                                        log.message.timestamp.difference.max. 
                                         ms                                   
                                        log.message.timestamp.type            
                                        log.preallocate                       
                                        log.retention.bytes                   
                                        log.retention.ms                      
                                        log.roll.jitter.ms                    
                                        log.roll.ms                           
                                        log.segment.bytes                     
                                        log.segment.delete.delay.ms           
                                        max.connection.creation.rate          
                                        max.connections                       
                                        max.connections.per.ip                
                                        max.connections.per.ip.overrides      
                                        message.max.bytes                     
                                        metric.reporters                      
                                        min.insync.replicas                   
                                        num.io.threads                        
                                        num.network.threads                   
                                        num.recovery.threads.per.data.dir     
                                        num.replica.fetchers                  
                                        principal.builder.class               
                                        replica.alter.log.dirs.io.max.bytes.  
                                         per.second                           
                                        sasl.enabled.mechanisms               
                                        sasl.jaas.config                      
                                        sasl.kerberos.kinit.cmd               
                                        sasl.kerberos.min.time.before.relogin 
                                        sasl.kerberos.principal.to.local.rules
                                        sasl.kerberos.service.name            
                                        sasl.kerberos.ticket.renew.jitter     
                                        sasl.kerberos.ticket.renew.window.    
                                         factor                               
                                        sasl.login.refresh.buffer.seconds     
                                        sasl.login.refresh.min.period.seconds 
                                        sasl.login.refresh.window.factor      
                                        sasl.login.refresh.window.jitter      
                                        sasl.mechanism.inter.broker.protocol  
                                        ssl.cipher.suites                     
                                        ssl.client.auth                       
                                        ssl.enabled.protocols                 
                                        ssl.endpoint.identification.algorithm 
                                        ssl.engine.factory.class              
                                        ssl.key.password                      
                                        ssl.keymanager.algorithm              
                                        ssl.keystore.certificate.chain        
                                        ssl.keystore.key                      
                                        ssl.keystore.location                 
                                        ssl.keystore.password                 
                                        ssl.keystore.type                     
                                        ssl.protocol                          
                                        ssl.provider                          
                                        ssl.secure.random.implementation      
                                        ssl.trustmanager.algorithm            
                                        ssl.truststore.certificates           
                                        ssl.truststore.location               
                                        ssl.truststore.password               
                                        ssl.truststore.type                   
                                        unclean.leader.election.enable        
                                       For entity-type 'users':               
                                        SCRAM-SHA-256                         
                                        SCRAM-SHA-512                         
                                        consumer_byte_rate                    
                                        controller_mutation_rate              
                                        producer_byte_rate                    
                                        request_percentage                    
                                       For entity-type 'clients':             
                                        consumer_byte_rate                    
                                        controller_mutation_rate              
                                        producer_byte_rate                    
                                        request_percentage                    
                                       Entity types 'users' and 'clients' may 
                                         be specified together to update      
                                         config for clients of a specific     
                                         user.

复制

修改 Broker 动态参数

修改动态参数无需重启 Broker，动态 Broker 参数的使用场景非常广泛，通常包括但不限于以下几种：

• 动态调整 Broker 端各种线程池大小，实时应对突发流量。
• 动态调整 Broker 端连接信息或安全配置信息。
• 动态更新 SSL Keystore 有效期。
• 动态调整 Broker 端 Compact 操作性能。
• 实时变更 JMX 指标收集器 (JMX Metrics Reporter)。

Kafka Broker Config 的参数有以下 3 种类型：

• read-only：被标记为 read-only 的参数和原来的参数行为一样，只有重启 Broker，才能令修改生效。
• per-broker：被标记为 per-broker 的参数属于动态参数，修改它之后，只会在对应的 Broker 上生效。
• cluster-wide：被标记为 cluster-wide 的参数也属于动态参数，修改它之后，会在整个集群范围内生效，也就是说，对所有 Broker 都生效。你也可以为具体的 Broker 修改 cluster-wide 参数。

在集群层面设置全局值，即设置 cluster-wide 范围值，将 ​​unclean.leader.election.enable​​ 参数在集群层面设置为 true。

kafka-configs.sh --bootstrap-server  10.37.249.58:9092 \
--entity-type brokers --entity-default --alter \
--add-config unclean.leader.election.enable=true

#返回结果
Completed updating default config for brokers in the cluster.

复制

如果要设置 cluster-wide 范围的动态参数，需要显式指定 entity-default。现在，我们使用下面的命令来查看一下刚才的配置是否成功。

kafka-configs.sh --bootstrap-server 10.37.249.58:9092 \
--entity-type brokers --entity-default --describe

#返回结果
Default configs for brokers in the cluster are:
  unclean.leader.election.enable=true sensitive=false synonyms={DYNAMIC_DEFAULT_BROKER_CONFIG:unclean.leader.election.enable=true}

复制

在 Zookeeper 上查看 /config/brokers/ 节点可以查看 cluster-wide 的动态参数设置。

[zk: (CONNECTED) ] > get /config/brokers/<default>
{"version":1,"config":{"unclean.leader.election.enable":"true"}}
cZxid = 17179869570
ctime = 1631246402937
mZxid = 17179869570
mtime = 1631246402937
pZxid = 17179869570
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0
dataLength = 64
numChildren = 0

复制

设置 per-broker 范围参数。我们还是以 ​​unclean.leader.election.enable​​ 参数为例，我现在为 ID 为 1 的 Broker 设置一个不同的值。命令如下：

kafka-configs.sh --bootstrap-server  10.37.249.58:9092 --entity-type brokers --entity-name 1 --alter --add-config unclean.leader.election.enable=false

#返回结果
Completed updating config for broker 1.

复制

我们使用下列命令查看 Broker ID 为 1 的节点动态参数，可以看到 ​​DYNAMIC_BROKER_CONFIG:unclean.leader.election.enable=false​​，表示我们刚才对 per-broker 参数的调整生效了。

kafka-configs.sh --bootstrap-server 10.37.249.58:9092 --entity-type brokers --entity-name 1 --describe

#返回结果
Dynamic configs for broker 1 are:
  unclean.leader.election.enable=false sensitive=false synonyms={DYNAMIC_BROKER_CONFIG:unclean.leader.election.enable=false, DYNAMIC_DEFAULT_BROKER_CONFIG:unclean.leader.election.enable=true, STATIC_BROKER_CONFIG:unclean.leader.election.enable=false, DEFAULT_CONFIG:unclean.leader.election.enable=false}

复制

在 Zookeeper 上查看 /config/brokers/1 节点可以查看 Broker ID 为 1 的节点的动态参数设置。

[zk: (CONNECTED) ] > get /config/brokers/1
{"version":1,"config":{"unclean.leader.election.enable":"false"}}
cZxid = 17179869574
ctime = 1631246495120
mZxid = 17179869574
mtime = 1631246495120
pZxid = 17179869574
cversion = 0
dataVersion = 0
aclVersion = 0
ephemeralOwner = 0
dataLength = 65
numChildren = 0[zk: (CONNECTED) ] > get /config/brokers/<default>[zk: (CONNECTED) ] > get /config/brokers/1

复制

删除 cluster-wide 范围动态参数。

kafka-configs.sh --bootstrap-server 10.37.249.58:9092 \
--entity-type brokers --entity-default --alter \
--delete-config unclean.leader.election.enable

#返回结果
Completed updating default config for brokers in the cluster.

复制

删除 per-broker 范围参数。

kafka-configs.sh --bootstrap-server 10.37.249.58:9092 \
--entity-type brokers --entity-name 1 --alter \
--delete-config unclean.leader.election.enable

#返回结果
Completed updating config for broker 1.

复制

修改 Topic 动态参数

设置 Topic test-topic 的 ​​retention.ms​​ 为 10000。

kafka-configs.sh --bootstrap-server  10.37.249.58:9092 \
--entity-type topics --entity-name test-topic --alter \
--add-config retention.ms=10000

复制

查看设置的 Topic 动态参数。

kafka-configs.sh --bootstrap-server  10.37.249.58:9092 \
--entity-type topics --entity-name test-topic --describe

#返回结果
Dynamic configs for topic test-topic are:
  retention.ms=10000 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=10000}

复制

在 Zookeeper 上可以查看 /config/topics/ 来查看 Topic 动态参数。

[zk: (CONNECTED) ] > get /config/topics/test-topic
{"version":1,"config":{"retention.ms":"10000"}}
cZxid = 17179869460
ctime = 1631245744105
mZxid = 17179869619
mtime = 1631250116481
pZxid = 17179869460
cversion = 0
dataVersion = 10
aclVersion = 0
ephemeralOwner = 0
dataLength = 47
numChildren = 0[zk: (CONNECTED) ] > get /config/topics/test-topic

复制

删除 Topic 动态参数。

kafka-configs.sh --bootstrap-server  10.37.249.58:9092 \
--entity-type topics --entity-name test-topic --alter \
--delete-config retention.ms

复制

Kafka 集群一键启动/停止脚本

环境变量设置：

#/etc/profile 文件
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin

复制

一键启动/停止脚本，查看状态需要安装 jps 工具。

#! /bin/bash
# 填写 Kafka Broker 节点地址
hosts=(kafka1 kafka2 kafka3)

# 打印启动分布式脚本信息
mill=`date "+%N"`
tdate=`date "+%Y-%m-%d %H:%M:%S,${mill:0:3}"`

echo [$tdate] INFO [Kafka Cluster] begins to execute the $1 operation.

# 执行分布式开启命令
function start()
{
        for i in ${hosts[@]}
                do
                        smill=`date "+%N"`
                        stdate=`date "+%Y-%m-%d %H:%M:%S,${smill:0:3}"`
                        ssh root@$i "source /etc/profile;echo [$stdate] INFO [Kafka Broker $i] begins to execute the startup operation.;kafka-server-start.sh $KAFKA_HOME/config/server.properties>/dev/null" &
                        sleep 1
                done
}

# 执行分布式关闭命令
function stop()
{
        for i in ${hosts[@]}
                do
                        smill=`date "+%N"`
                        stdate=`date "+%Y-%m-%d %H:%M:%S,${smill:0:3}"`
                        ssh root@$i "source /etc/profile;echo [$stdate] INFO [Kafka Broker $i] begins to execute the shutdown operation.;kafka-server-stop.sh>/dev/null;" &
                        sleep 1
                done
}

# 查看 Kafka Broker 节点状态
function status()
{
        for i in ${hosts[@]}
                do
                        smill=`date "+%N"`
                        stdate=`date "+%Y-%m-%d %H:%M:%S,${smill:0:3}"`
                        ssh root@$i "source /etc/profile;echo [$stdate] INFO [Kafka Broker $i] status message is :;jps | grep Kafka;" &
                        sleep 1
                done
}

# 判断输入的 Kafka 命令参数是否有效
case "$1" in
    start)
        start
        ;;
    stop)
        stop
        ;;
    status)
        status
        ;;
    *)
        echo "Usage: $0 {start|stop|status}"
        RETVAL=1
esac

复制

参考资料

• Kafka 动态配置了解下?(https://time.geekbang.org/column/article/113504)
• 常见工具脚本大汇总(https://time.geekbang.org/column/article/116111)
• Kafka 并不难学！