Kafka：如何使用JavaAPI从主题中删除记录？

Question

我正在寻找一种从Kafka主题中删除（完全删除）已消费记录的方法。我知道有几种方法可以做到这一点，例如更改主题的保留时间或删除Kafka logs文件夹。但我要寻找的是一种使用Java API删除某个主题的一定数量记录的方法，如果可能的话。

我试过测试AdminClient API，特别是AdminClient。deleteRecords（recordsToDelete）方法。但如果我没弄错的话，这种方法只会改变主题中的偏移量，而不是从硬盘上删除所述记录。

有没有一个Java API可以从硬盘上删除记录？

Answer 1

Kafka主题是不可变的，这意味着您只能向它们添加新消息。没有删除本身。

然而，为了避免“磁盘耗尽”，Kafka提供了两个概念来降低主题的大小：保留策略和压缩。

保留如果你有一个你永远不需要数据的主题，你只需设置一个保留策略，不管你需要多长时间保留数据，即72小时。Kafka会自动删除超过72小时的邮件。

压缩如果您确实需要数据永远存在，或者至少长时间存在，但是您只需要最新的值，那么您可以将主题设置为压缩。一旦新消息添加了一个已经存在的密钥，这个将自动删除旧消息。

规划Kafka架构的核心部分是思考如何将数据存储在主题中。例如，如果你在Kafka主题中推送客户记录的更新，比如说客户的最后登录日期（非常做作的例子…），然后你只对最后一个条目感兴趣（因为之前的所有条目都不再是“最后”登录）。如果此分区的密钥是客户ID，并且启用了日志压缩，那么一旦用户登录并且kafka主题收到此事件，具有相同分区密钥（客户ID）的任何其他先前消息都将自动从主题中删除。

Answer 2

我可以删除。如果linux在机器上，它会将其从硬盘中删除。当我在网上搜索时，我发现windows有一个漏洞。然而，我在windows中找不到解决这个错误的方法。如果kafka在linux机器上运行，则此代码有效。

public void deleteMessages(String topicName, int partitionIndex, int beforeIndex) {
       TopicPartition topicPartition = new TopicPartition(topicName, partitionIndex);
       Map<TopicPartition, RecordsToDelete> deleteMap = new HashMap<>();
       deleteMap.put(topicPartition, RecordsToDelete.beforeOffset(beforeIndex));
       kafkaAdminClient.deleteRecords(deleteMap);
}

Answer 3

这让我一开始也有点困惑，为什么包括bin/kafka删除记录。sh可以删除，但我不能使用Java API

缺少的是你需要给Kafka夫未来打电话。get（），因为deleteRecords返回未来的映射

这是密码

在这段代码中，您需要调用entry.getValue（）. get（）. lowWatermark（）

DeleteRecordsResult result = adminClient.deleteRecords(recordsToDelete);
Map<TopicPartition, KafkaFuture<DeletedRecords>> lowWatermarks = result.lowWatermarks();
try {
    for (Map.Entry<TopicPartition, KafkaFuture<DeletedRecords>> entry : lowWatermarks.entrySet()) {
        System.out.println(entry.getKey().topic() + " " + entry.getKey().partition() + " " + entry.getValue().get().lowWatermark());
    }
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
}
adminClient.close();