如何删除消费者已经消费过的数据？卡夫卡

我正在kafka中进行数据复制。但是，kafka 日志文件的大小增长得非常快。一天的大小达到 5 GB。作为这个问题的解决方案，我想立即删除已处理的数据。我正在 AdminClient 中使用删除记录方法来删除偏移量。但是当我查看日志文件时，与该偏移量对应的数据并未被删除。

RecordsToDelete recordsToDelete = RedcordsToDelete.beforeOffset(offset);
TopicPartition topicPartition = new TopicPartition(topicName,partition);
Map<TopicPartition,RecordsToDelete> deleteConf = new HashMap<>();
deleteConf.put(topicPartition,recordsToDelete);
adminClient.deleteRecords(deleteConf);

我不需要像这样的建议（log.retention.hours、log.retention.bytes、log.segment.bytes、log.cleanup.policy=删除）

因为我只想删除消费者消费的数据。在这个解决方案中，我还删除了没有消耗的数据。

您有什么建议？

你没有做错任何事。您提供的代码有效并且我已经测试过。以防万一我忽略了您代码中的某些内容，我的代码是：

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);
}

我使用过组：'org.apache.kafka'，名称：'kafka-clients'，版本：'2.0.0'

因此，请检查您的目标分区是否正确（第一个分区为 0）

检查您的经纪人版本：https://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/admin/AdminClient.html https://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/admin/AdminClient.html says:

0.11.0.0版本的经纪商支持此操作

从同一应用程序生成消息，以确保连接正确。

您还可以考虑另一种选择。使用cleanup.policy=紧凑如果您的消息键重复，您可能会从中受益。不仅因为该键的较旧消息将被自动删除，而且您可以利用具有空负载的消息删除该键的所有消息这一事实。只是不要忘记设置删除.retention.ms and 最小压缩延迟毫秒到足够小的值。在这种情况下，您可以使用消息，然后为同一密钥生成空有效负载（但要谨慎使用此方法，因为这样您可以删除未使用的消息（使用该密钥））