零壹

01

Kafka Notes

发表于 更新于 分类于
本文字数： 2.3k 阅读时长 ≈ 2 分钟

简介

…

部署

Docker

172.16.18.143 为宿主机 IP

  • Zookeeper

    1
    2
    docker run --name eventbus_zk -d -p 2181:2181 zookeeper:3.5.7
    # docker run --name eventbus_zk -d -p 2181:2181 wurstmeister/zookeeper:3.4.6

  • Kafka

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    docker run --name eventbus_kafka -d \
       -p 9092:9092 \
       -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 \
       -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://172.16.18.143:9092 \
       -e KAFKA_ZOOKEEPER_CONNECT=172.16.18.143:2181 \
       -e KAFKA_AUTO_CREATE_TOPICS_ENABLE=true \
       -e KAFKA_DELETE_TOPIC_ENABLE=true \
       -e KAFKA_BROKER_ID=0 \
       -v /var/run/docker.sock:/var/run/docker.sock \
       wurstmeister/kafka:2.11-1.1.1

命令行

示例

  • Topic List

    1
    /opt/kafka/bin/kafka-topics.sh --list --zookeeper 172.16.18.143:2181

  • Producer

    1
    /opt/kafka/bin/kafka-console-producer.sh --broker-list 172.16.18.143:9092 --topic test.eb1

  • Consumer

    1
    /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server 172.16.18.143:9092 --topic test.eb1  --from-beginning

运维日志

清除 Topic 数据（不删除主题）

思路：

  1. 更改 topic 配置，指定数据保留时间retension.ms=1000为1秒（默认7天）；指定清除策略为cleanup.policy=delete
  2. 查看 topic 配置，确保配置生效
  3. 等待 kafka 执行清理操作，并确认 topic 数据被清理。
  4. 恢复 topic 配置，删除 retension.ms 配置项，更改清除策略为 cleanup.policy=compact。

示例：删除 topic connect-status,connect-config,connect-offsets数据。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
# connect-status
kafka-configs --zookeeper 172.16.0.61:2181 --alter --entity-type topics --add-config retention.ms=1000,cleanup.policy=delete --entity-name connect-status
kafka-configs --zookeeper 172.16.0.61:2181 --entity-type topics --describe --entity-name connect-status
kafka-configs --zookeeper 172.16.0.61:2181  --alter --entity-type topics --add-config cleanup.policy=compact --delete-config retention.ms --entity-name connect-status

# connect-config
kafka-configs --zookeeper 172.16.0.61:2181 --alter --entity-type topics --add-config retention.ms=1000,cleanup.policy=delete --entity-name connect-config
kafka-configs --zookeeper 172.16.0.61:2181 --entity-type topics --describe --entity-name connect-config
kafka-configs --zookeeper 172.16.0.61:2181  --alter --entity-type topics --add-config cleanup.policy=compact --delete-config retention.ms --entity-name connect-config

# connect-offsets
kafka-configs --zookeeper 172.16.0.61:2181 --alter --entity-type topics --add-config retention.ms=1000,cleanup.policy=delete --entity-name connect-offsets
kafka-configs --zookeeper 172.16.0.61:2181 --entity-type topics --describe --entity-name connect-offsets
kafka-configs --zookeeper 172.16.0.61:2181 --alter --entity-type topics --add-config cleanup.policy=compact --delete-config retention.ms --entity-name connect-offsets

（待续）