我在Kafka做数据复制。但是,kafka日志文件的大小增长很快。一天的大小达到5GB。为了解决这个问题,ı 要立即删除已处理的数据。我在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)
因为我只想删除消费者消费的数据。在这个解决方案中,我还删除了未使用的数据。
你有什么建议?
2条答案
按热度按时间js4nwp541#
你没做错什么。你提供的代码工作正常,我已经测试过了。万一我忽略了你代码中的某些内容,我的代码是:
我使用了group:'org.apache.kafka',name:'kafka clients',version:'2.0.0'
因此,请检查是否以正确的分区为目标(第一个分区为0)
检查您的代理版本:https://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/admin/adminclient.html 说:
版本为0.11.0.0的代理支持此操作
从同一个应用程序生成消息,以确保连接正确。
还有一个你可以考虑的选择。使用cleanup.policy=compact如果您的消息键是重复的,您可以从中受益。不仅因为该键的旧消息将被自动删除,而且您可以使用这样一个事实,即负载为null的消息将删除该键的所有消息。只是别忘了将delete.retention.ms和min.compression.lag.ms的值设置得足够小。在这种情况下,您可以使用一条消息,然后为同一个键生成空负载(但是要谨慎使用这种方法,因为这样可以删除(使用该键)您没有使用的消息)
qxgroojn2#
试试这个
在这段代码中,您需要调用
entry.getValue().get().lowWatermark()
,因为adminclient.deleterecords(recordstodelete)返回未来的Map,所以需要通过调用get()等待未来运行