一、创建/删除/重建topic

1、创建：./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 2 --partitions 10 --topic TOPIC_NAME

2、删除：./bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic TOPIC_NAME

3、查看状态和分区负载详情：./bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic TOPIC_NAME

二、设置过期时间

以下例子为保存7天

处理方法：

1、设置：./bin/kafka-configs.sh --zookeeper localhost:2181  --entity-type topics --alter --entity-name TOPIC_NAME --add-config retention.ms=604800000

2、删除：./bin/kafka-configs.sh --zookeeper localhost:2181  --entity-type topics --alter --entity-name TOPIC_NAME --delete-config retention.ms

三、删除某个offset之前的数据

说明：需求是删除某个offset之前的数据；一般来说，直接重建topic比较快，但是不排除不能重建topic的情况

处理方法：

1、创建json文件，以下例子是删除topic：test_product，分区：0，offset：10之前的数据，不包含offset 10，改操作不会重置offset

示例：{"partitions":[{"topic": "test_product", "partition": 0,"offset": 10}],"version":1}

2、执行命令：./bin/kafka-delete-records.sh --bootstrap-server localhost:9092 --offset-json-file cleanup.json

四、kafka磁盘报警

处理方法：

1、查看哪个topic数据量比较大，进入kafka数据目录，执行命令查看分区数大于1G的：du -sh * |grep G （注意：不一定是topic数据量大，也有可能是日志之类）

2、查看数据量最大的topic是否设置过期时间，如果未设置，和业务部门确认是否可以设置；如果已设置，是否需要缩短时间或者重建topic

3、如果以上方法都无法解决，让业务部门提单扩容磁盘。

五、kafka_exporter报警

处理方法：一般情况是CPU使用率过高导致拉取不到监控数据，也有可能是网络问题；查看监控面板，如果CPU使用率突然升高需要确认，如果CPU持续较高需要升级配置。

六、重置消费者offset

说明：有些场景可能希望修改消费者消费到的offset位置，以达到重新消费，或者跳过一部分消息的目的

处理方法：

1、将指定GROUP_NAME和topic的offset修改到NEW_OFFSET的位置，重启消费者后，消费中将从指定的offset处消费。

注意这里只能NEW_OFFSET只能设置一个值，也就是说，所有的分区都将使用这个值，如果分区消息负载不均衡，需要考虑是否适用。

./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group GROUP_NAME --reset-offsets --execute --to-offset NEW_OFFSET --topic TOPIC_NAME

2、将指定GROUP_NAME和topic的offset修改到earliest或者latest位置，使得消费者从头或者从尾部消费。

./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group GROUP_NAME --reset-offsets --execute --to-earliest --topic TOPIC_NAME（从头消费）

./bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group GROUP_NAME --reset-offsets --execute --to-latest --topic TOPIC_NAME（从最新消费）

七、查询topic的offset的范围

说明：一般很少用到，可以直接在kafdrop上面看，如果没有kafdrop的时候使用以下方法查询

处理方法：

查询offset的最小值（理解为消费者的offset）

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 -topic TOPIC_NAME --time -2

查询offset的最大值（理解为生产者的offset）

./bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 -topic TOPIC_NAME --time -1

八、导出某个topic的数据

说明：注意topic数据量大小和磁盘大小，最好不要重定向到根分区；如果需要下载先压缩再下载，压缩比很高

处理方法：从头开始消费重定向到文件

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

九、增加topic分区数（只能加不能减）

# ./bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 -alter --partitions 12 --topic ec-applog-err
WARNING: If partitions are increased for a topic that has a key, the partition logic or ordering of the messages will be affected
Adding partitions succeeded!

# 这里给了一个警告 : 如果为具有关键字的主题增加了分区，则分区逻辑或消息的顺序将受到影响。
# 如果我们的业务场景对消息的顺序有着严格的要求, 一定要谨慎添加分区! 建议将之前所有的消息全部消费完才执行添加分区操作

十、控制台生产者/消费者

生产者：./bin/kafka-console-producer--broker-list localhost:9092 --topic test1

消费者：./bin/kafka-console-consumer--zookeeper localhost:2181 --topic test1 --from-beginning 

十一、停止/启动服务

启动：./bin/kafka-server-start.sh -daemon config/server.properties

停止：./bin/kafka-server-stop.sh

十二、kafka鉴权配置

配置文件示例server.properties

# -----------  security begin ----------- authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer super.users=User:admin sasl.enabled.mechanisms=SCRAM-SHA-512 sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512 security.inter.broker.protocol=SASL_PLAINTEXT listeners=SASL_PLAINTEXT://192.168.1.213:9092 advertised.listeners=SASL_PLAINTEXT://192.168.1.213:9092 # -----------  security end -----------

配置文件示例kafka_server_jaas.conf

KafkaServer {    org.apache.kafka.common.security.scram.ScramLoginModule required    username="admin"    password="admin-secret"; };

修改启动脚本，添加bin/kafka-server-start.sh

export KAFKA_OPTS=-Djava.security.auth.login.config=$KAFKA_HOME/config/jaas.conf

授权脚本

[root@data-kafka-auth-prod01 scripts]# cat chmod_read.sh #!/bin/sh #通过传入用户名、用户消费组作为参数、即可批量赋权 cat topic | while read line do   kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:$1 --operation Read --group $2 --topic $line done [root@data-kafka-auth-prod01 scripts]# cat chmod_write.sh #!/bin/sh #通过传入用户名、即可批量赋写权限 cat topic | while read line do   kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:$1 --operation Write --topic $line done