我正在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 最小压缩延迟毫秒到足够小的值。在这种情况下,您可以使用消息,然后为同一密钥生成空有效负载(但要谨慎使用此方法,因为这样您可以删除未使用的消息(使用该密钥))
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)