有没有办法清除Kafka中的主题？

Question

我在本地计算机上将一条太大的消息推送到Kafka消息主题中，现在我收到一个错误：

kafka.common.InvalidMessageSizeException: invalid message size

增加fetch.size在这里并不理想，因为我实际上不想接受那么大的消息。

Answer 1

虽然公认的答案是正确的，但该方法已被弃用。主题配置现在应该通过kafka configs完成。

kafka-configs --zookeeper localhost:2181 --entity-type topics --alter --add-config retention.ms=1000 --entity-name MyTopic

通过此方法设置的配置可以通过命令显示

kafka-configs --zookeeper localhost:2181 --entity-type topics --describe --entity-name MyTopic

Answer 2

要清除队列，您可以删除主题：

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic test

然后重新创建：

bin/kafka-topics.sh --create --zookeeper localhost:2181 \
    --replication-factor 1 --partitions 1 --topic test

Answer 3

暂时将主题的保留时间更新为一秒：

kafka-topics.sh \
  --zookeeper <zkhost>:2181 \
  --alter \
  --topic <topic name> \
  --config retention.ms=1000

在较新的Kafka版本中，您还可以使用Kafka配置--实体类型主题

kafka-configs.sh \
  --zookeeper <zkhost>:2181 \
  --entity-type topics \
  --alter \
  --entity-name <topic name> \
  --add-config retention.ms=1000

然后等待清除生效（持续时间取决于主题的大小）。清除后，恢复以前的保留。ms值。