调用deleteRecords Kafka Admin Client Java API时，不会从文件系统中删除消息

Question

我试图使用Java Admin Client API的delete Records方法从我的kafka主题中删除消息。以下是我尝试的步骤

    1. I pushed 20000 records to my TEST-DELETE topic
    2. Started a console consumer and consumed all the messages
    3. Invoked my java program to delete all those 20k messages
    4. Started another console consumer with a different group id. This consumer is not receiving any of the deleted messages

当我检查文件系统时，我仍然可以看到占用磁盘空间的所有20K记录。我的意图是删除这些记录永远从文件系统太。

Topic:TEST-DELETE       PartitionCount:4        ReplicationFactor:1     Configs:cleanup.policy=delete
        Topic: TEST-DELETE    Partition: 0      Leader: 0     Replicas: 0       Isr: 0
        Topic: TEST-DELETE    Partition: 1      Leader: 0     Replicas: 0       Isr: 0
        Topic: TEST-DELETE    Partition: 2      Leader: 0     Replicas: 0       Isr: 0
        Topic: TEST-DELETE    Partition: 3      Leader: 0     Replicas: 0       Isr: 0

    log.retention.hours=24
    log.retention.check.interval.ms=60000
    log.cleaner.delete.retention.ms=60000
    file.delete.delay.ms=60000
    delete.retention.ms=60000
    offsets.retention.minutes=5
    offsets.retention.check.interval.ms=60000
    log.cleaner.enable=true
    log.cleanup.policy=compact,delete

下面给出了我的删除代码

public void deleteRecords(Map<String, Map<Integer, Long>> allTopicPartions) {

        Map<TopicPartition, RecordsToDelete> recordsToDelete = new HashMap<>();

        allTopicPartions.entrySet().forEach(topicDetails -> {

            String topicName = topicDetails.getKey();
            Map<Integer, Long> value = topicDetails.getValue();

            value.entrySet().forEach(partitionDetails -> {

                if (partitionDetails.getValue() != 0) {
                    recordsToDelete.put(new TopicPartition(topicName, partitionDetails.getKey()),
                            RecordsToDelete.beforeOffset(partitionDetails.getValue()));
                }
            });
        });

        DeleteRecordsResult deleteRecords = this.client.deleteRecords(recordsToDelete);

        Map<TopicPartition, KafkaFuture<DeletedRecords>> lowWatermarks = deleteRecords.lowWatermarks();

        lowWatermarks.entrySet().forEach(entry -> {
            try {
                logger.info(entry.getKey().topic() + " " + entry.getKey().partition() + " "
                        + entry.getValue().get().lowWatermark());
            } catch (Exception ex) {

            }
        });

    }

下面给出了我的java程序的输出

2019-06-25 16:21:15 INFO  MyKafkaAdminClient:247 - TEST-DELETE 1 5000
2019-06-25 16:21:15 INFO  MyKafkaAdminClient:247 - TEST-DELETE 0 5000
2019-06-25 16:21:15 INFO  MyKafkaAdminClient:247 - TEST-DELETE 3 5000
2019-06-25 16:21:15 INFO  MyKafkaAdminClient:247 - TEST-DELETE 2 5000

我的意图是从文件系统中删除消耗的记录，因为我的kafka代理的存储空间有限。

null

Answer 1

处理此问题的建议方法是为您感兴趣的主题设置retention.ms和相关配置值。这样，您就可以定义Kafka在删除数据之前将存储数据的时间，以确保在数据从Kafk集群中删除之前，您的所有下游消费者都有机会下拉数据。

但是，如果您仍然希望强制Kafka基于字节进行删除，则有log.retention.bytes和retention.bytes配置值。第一个是集群范围的设置，第二个是特定于主题的设置，默认情况下，它接受第一个设置的任何设置，但您仍然可以针对每个主题重写它。retention.bytes数字是每个分区强制执行的，因此应该将其乘以主题分区的总数。

但是，请注意，如果有一个失控的生产者突然开始生成大量数据，并且将其设置为硬字节限制，则可能会清除集群中一整天的数据，只剩下最后几分钟的数据，甚至可能在有效的使用者从集群中拉下数据之前。这就是为什么将Kafka主题设置为基于时间的保留要好得多，而不是基于字节的。

您可以在官方的Kafka文档中找到配置属性及其解释：https://Kafka.apache.org/documentation/