无需设置保留天数清除Kafka Topic：求其他清空方案及示例

嘿，我来分享几个除了删除Topic或者修改保留天数之外，能清空Kafka Topic的实用方法，每个都带操作示例，方便你上手：

这是官方提供的最直接的物理清空方式，能精准删除指定分区内某偏移量之前的所有消息，适合需要彻底清除数据的场景。

操作步骤：

1. 先创建一个JSON配置文件（比如delete-topic-config.json），指定要清空的Topic和分区，把偏移量设为-1（代表删除到最早的偏移量之前，也就是清空该分区所有消息）：

{
  "partitions": [
    {
      "topic": "my-target-topic",
      "partition": 0,
      "offset": -1
    },
    {
      "topic": "my-target-topic",
      "partition": 1,
      "offset": -1
    }
  ],
  "version": 1
}

如果Topic有多个分区，把所有分区都列进去就行。

2. 运行工具命令执行删除：

kafka-delete-records.sh --bootstrap-server your-kafka-broker:9092 --offset-json-file delete-topic-config.json

执行成功后，你会收到Broker返回的确认信息，说明指定分区的消息已经被物理删除。

这个方法不是物理删除消息，而是让指定消费组的偏移量直接跳到最新位置，这样该消费组后续就看不到旧消息了。适合不想改动Broker配置，只是想让某个消费组重新开始消费的场景。

命令行快速实现：
用kafka自带的控制台消费者，从头消费所有消息并自动提交偏移量，输出直接丢到/dev/null：

kafka-console-consumer.sh --bootstrap-server your-kafka-broker:9092 --topic my-target-topic --group cleanup-consumer-group --from-beginning --consumer-property enable.auto.commit=true --consumer-property auto.commit.interval.ms=1000 > /dev/null

注意：这个操作只对cleanup-consumer-group这个消费组生效，其他消费组如果偏移量没更新，还是能看到旧消息。

Java代码实现（适合自动化）：

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class TopicCleanupConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-broker:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "cleanup-consumer-group");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(Collections.singletonList("my-target-topic"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                if (records.isEmpty()) {
                    // 没有剩余消息，提交最新偏移量
                    consumer.commitSync();
                    break;
                }
                // 无需处理消息，直接提交当前偏移量
                consumer.commitSync();
            }
            System.out.println("消费组偏移量已更新，旧消息不再可见");
        }
    }
}

如果需要在代码里自动完成清空操作，可以用Kafka的AdminClient来调用删除接口，原理和kafka-delete-records工具一样，都是物理删除消息。

Java代码示例：

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.TopicPartition;
import java.util.*;
import java.util.concurrent.ExecutionException;

public class TopicCleanupAdmin {
    public static void main(String[] args) {
        Properties adminProps = new Properties();
        adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "your-kafka-broker:9092");

        try (AdminClient adminClient = AdminClient.create(adminProps)) {
            // 获取目标Topic的所有分区
            DescribeTopicsResult describeResult = adminClient.describeTopics(Collections.singletonList("my-target-topic"));
            TopicDescription topicDesc = describeResult.values().get("my-target-topic").get();
            
            List<TopicPartition> partitions = new ArrayList<>();
            for (TopicPartitionInfo p : topicDesc.partitions()) {
                partitions.add(new TopicPartition("my-target-topic", p.partition()));
            }

            // 获取每个分区的最新偏移量，设置删除到该偏移量（即清空所有旧消息）
            Map<TopicPartition, OffsetSpec> offsetSpecs = new HashMap<>();
            for (TopicPartition tp : partitions) {
                offsetSpecs.put(tp, OffsetSpec.latest());
            }
            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> offsetResults = adminClient.listOffsets(offsetSpecs).all().get();
            
            // 执行删除操作
            DeleteRecordsResult deleteResult = adminClient.deleteRecords(
                offsetResults.entrySet().stream()
                    .collect(Map::ofEntries,
                        (m, e) -> m.put(e.getKey(), new DeleteRecordsRequestData.DeletePartition(e.getValue().offset())),
                        Map::putAll)
            );
            deleteResult.all().get();
            
            System.out.println("Topic已成功清空");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

如果你的Topic设置了cleanup.policy=compact（压缩策略），可以通过发送墓碑消息（tombstone）触发压缩，从而清理掉旧的消息。这个方法适合需要保留Topic结构，但要清空历史数据的压缩Topic。

操作步骤：

1. 确保Topic的清理策略是compact（如果没设置，先修改）：

kafka-configs.sh --bootstrap-server your-kafka-broker:9092 --alter --topic my-compacted-topic --add-config cleanup.policy=compact

2. 发送一条墓碑消息（键为你要清空的消息的键，值为空）：

kafka-console-producer.sh --bootstrap-server your-kafka-broker:9092 --topic my-compacted-topic --property parse.key=true --property key.separator=:
# 在控制台输入：my-key: （注意冒号后为空，代表值为null）

3. 加快压缩触发（可选，默认压缩会自动触发，修改比例可以让它更快执行）：

kafka-configs.sh --bootstrap-server your-kafka-broker:9092 --alter --topic my-compacted-topic --add-config log.cleaner.min.cleanable.ratio=0.01

压缩完成后，所有对应键的旧消息都会被清理掉。如果Topic里的消息用的是同一个键，这就能达到清空的效果。

内容的提问来源于stack exchange，提问作者Ronnie Phelps