常用命令

Broker 命令

kafka协调器的问题

#查看消费情况
bin/kafka-topics.sh -bootstrap-server 10.10.1.17:9092 --topic __consumer_offsets --describe
#查看kafka的leader的副本情况
cat config/server.properties | grep offsets.topic.replication.factor

启动Kafka服务器

bin/kafka-server-start.sh -daemon  config/server.properties

停止Kafka服务器

bin/kafka-server-stop.sh

创建主题（Topic）

bin/kafka-topics.sh --create --topic test --bootstrap-server master1:9092,master2:9092,node1:9092 --partitions <num-partitions> --replication-factor <replication-factor>

查看所有的消费者组信息

bin/kafka-consumer-groups.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --list

查看主题列表

bin/kafka-topics.sh --list --bootstrap-server master1:9092,master2:9092,node1:9092

查看主题详情

bin/kafka-topics.sh --describe --topic test --bootstrap-server master1:9092,master2:9092,node1:9092

查看消费者组信息

bin/kafka-consumer-groups.sh --describe --group testgroup --bootstrap-server master1:9092,master2:9092,node1:9092

查看所有消费者组消费情况

bin/kafka-consumer-groups.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --all-groups --describe

查看Broker状态

bin/kafka-broker-api-versions.sh --bootstrap-server master1:9092,master2:9092,node1:9092

查看Kafka日志

tail -f logs/server.log

设置消费者组的位置(offset)偏移

设置为最新的消费

bin/kafka-consumer-groups.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --group testgroup --reset-offsets --to-latest --execute --topic test1

设置为最早的位置消费

bin/kafka-consumer-groups.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --group testgroup --reset-offsets --to-earliest --execute --topic test1

由于消费者指定了--group testgroup消费以后，后面即使使用了--from-beginning也会从最新的地方消费，因为只有在存储offset主题里面的消费者组是最新的时候，--from-beginning才会生效。使用上面的命令可以手动的设置消费的位置。

查看消费者组对应的偏移量情况

bin/kafka-consumer-groups.sh --describe --group testgroup --bootstrap-server master1:9092,master2:9092,node1:9092

查看Kafka的版本信息

bin/kafka-broker-api-versions.sh --bootstrap-server master1:9092,master2:9092,node1:9092

Product 命令

向主题发送消息

bin/kafka-console-producer.sh --topic test1 --bootstrap-server master1:9092,master2:9092,node1:9092

Consumer 命令

从主题消费消息

bin/kafka-console-consumer.sh --topic test1 --bootstrap-server master1:9092,master2:9092,node1:9092 --group testgroup --from-beginning

有时候我们发现 Topic 的 leader 不是我们想要的顺序我们应该怎么处理？

比如下面的情况，怎么才能让leader分布均匀一点？

bin/kafka-topics.sh --describe --topic test --bootstrap-server master1:9092,master2:9092,node1:9092

[bigdata@node1 kafka_2.12-3.6.0]$ bin/kafka-topics.sh --describe --topic test --bootstrap-server master1:9092,master2:9092,node1:9092
Topic: test     TopicId: 9eanKgjfQDaIS1yavdhnRQ PartitionCount: 5       ReplicationFactor: 3    Configs: flush.ms=1000,segment.bytes=1073741824,flush.messages=10000,max.message.bytes=5242880
        Topic: test     Partition: 0    Leader: 0       Replicas: 1,2,0 Isr: 1,0,2
        Topic: test     Partition: 1    Leader: 0       Replicas: 2,0,1 Isr: 1,0,2
        Topic: test     Partition: 2    Leader: 1       Replicas: 1,2,0 Isr: 1,0,2
        Topic: test     Partition: 3    Leader: 0       Replicas: 0,2,1 Isr: 1,0,2
        Topic: test     Partition: 4    Leader: 1       Replicas: 2,1,0 Isr: 1,0,2

方案一

手动自定义分区分配。

vi reassignment.json

{
  "partitions": [
    {"topic": "test", "partition": 0, "replicas": [1, 2, 0]},
    {"topic": "test", "partition": 1, "replicas": [2, 0, 1]},
    {"topic": "test", "partition": 2, "replicas": [1, 2, 0]},
    {"topic": "test", "partition": 3, "replicas": [0, 2, 1]},
    {"topic": "test", "partition": 4, "replicas": [2, 1, 0]}
  ]
}

执行下面的命令

bin/kafka-reassign-partitions.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --reassignment-json-file reassignment.json --execute

如下leader分配均匀了

[bigdata@node1 kafka_2.12-3.6.0]$ bin/kafka-topics.sh --describe --topic test --bootstrap-server master1:9092,master2:9092,node1:9092
Topic: test     TopicId: 9eanKgjfQDaIS1yavdhnRQ PartitionCount: 5       ReplicationFactor: 3    Configs: flush.ms=1000,segment.bytes=1073741824,flush.messages=10000,max.message.bytes=5242880
        Topic: test     Partition: 0    Leader: 0       Replicas: 1,2,0 Isr: 1,0,2
        Topic: test     Partition: 1    Leader: 2       Replicas: 2,0,1 Isr: 1,0,2
        Topic: test     Partition: 2    Leader: 1       Replicas: 1,2,0 Isr: 1,0,2
        Topic: test     Partition: 3    Leader: 0       Replicas: 0,2,1 Isr: 1,0,2
        Topic: test     Partition: 4    Leader: 2       Replicas: 2,1,0 Isr: 1,0,2

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

sh bin/kafka-leader-election.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --topic test --election-type PREFERRED --partition 0

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

sh bin/kafka-leader-election.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --election-type preferred  --all-topic-partitions

方案二

bin/kafka-topics.sh --describe --topic test --bootstrap-server master1:9092,master2:9092,node1:9092

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

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

sh bin/kafka-leader-election.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --topic test --election-type PREFERRED --partition 0

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

sh bin/kafka-leader-election.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --election-type preferred  --all-topic-partitions

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

原始的情况

        Topic: test     Partition: 0    Leader: 1       Replicas: 1,2,0 Isr: 0,1,2
        Topic: test     Partition: 1    Leader: 0       Replicas: 2,0,1 Isr: 0,1,2
        Topic: test     Partition: 2    Leader: 1       Replicas: 1,2,0 Isr: 0,1,2
        Topic: test     Partition: 3    Leader: 0       Replicas: 0,2,1 Isr: 0,1,2
        Topic: test     Partition: 4    Leader: 1       Replicas: 2,1,0 Isr: 0,1,2

vi config/leader-election.json

{
  "partitions": [
    {
      "topic": "test",
      "partition": 0
    },
    {
      "topic": "test",
      "partition": 1
    },
    {
      "topic": "test",
      "partition": 2
    },
    {
      "topic": "test",
      "partition": 3
    },
    {
      "topic": "test",
      "partition": 4
    }
  ]
}

sh bin/kafka-leader-election.sh --bootstrap-server master1:9092,master2:9092,node1:9092 --election-type preferred  --path-to-json-file config/leader-election.json

重新选举leader以后

        Topic: test     Partition: 0    Leader: 1       Replicas: 1,2,0 Isr: 0,1,2
        Topic: test     Partition: 1    Leader: 2       Replicas: 2,0,1 Isr: 0,1,2
        Topic: test     Partition: 2    Leader: 1       Replicas: 1,2,0 Isr: 0,1,2
        Topic: test     Partition: 3    Leader: 0       Replicas: 0,2,1 Isr: 0,1,2
        Topic: test     Partition: 4    Leader: 2       Replicas: 2,1,0 Isr: 0,1,2

Broker 命令​

kafka协调器的问题​

启动Kafka服务器​

停止Kafka服务器​

创建主题（Topic）​

查看所有的消费者组信息​

查看主题列表​

查看主题详情​

查看消费者组信息​

查看所有消费者组消费情况​

查看Broker状态​

查看Kafka日志​

设置消费者组的位置(offset)偏移​

查看Kafka的版本信息​

Product 命令​

向主题发送消息​

Consumer 命令​

从主题消费消息​

有时候我们发现 Topic 的 leader 不是我们想要的顺序我们应该怎么处理？​

方案一​

方案二​