如何删除消费者已经消费的数据？Kafka

Question

我在Kafka做数据复制。但是，kafka日志文件的大小增长很快。一天内大小达到5 gb。作为这个问题解决方案，我想立即删除处理过的数据。我正在使用AdminClient中的delete record方法删除偏移量。但当我查看日志文件时,对应于该偏移量的数据不会被删除。

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=delete)的建议

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

你有什么建议？

Answer 1

你没做错什么。您提供的代码可以工作，我已经测试过了。以防万一我忽略了您的代码中的某些内容，我的代码是：

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表示第一个分区）

还有一个你可以考虑的选择。使用cleanup.policy=compact如果您的消息键重复，您可以从中受益。不仅仅是因为该密钥的旧消息会自动删除，还可以利用空负载的消息会删除该密钥的所有消息这一事实。只是不要忘记将delete.retention.ms和min.compaction.lag.ms设置为足够小的值。在这种情况下，您可以使用一个消息，然后为相同的密钥生成空有效负载（但是使用这种方法要谨慎，因为这样您可以删除您没有使用的消息（使用该密钥））