如何解决删除后 Kafka KeyValue 存储的磁盘空间未更新
我的服务正在使用 Kafka 键值存储。我已经在两个处理器中安排了两个任务,它们从存储中删除格式错误和过时的记录。但是,我只有在停止本地实例后才能看到更新的磁盘空间。我查了一下,从我的 java 进程到 RocksDB 有很多句柄,不管任务是否正在运行。
这是我的代码:
public void init(ProcessorContext processorContext) {
store = (keyvalueStore<String,AssistantMessage>) processorContext.getStateStore(storeName);
processorContext.schedule(Duration.ofHours(6),PunctuationType.WALL_CLOCK_TIME,timestamp -> {
LOGGER.info("Starting execution of task for deletion of assistant group messages.");
try( keyvalueIterator<String,AssistantMessage> iterator = store.all()) {
while (iterator.hasNext()) {
keyvalue<String,AssistantMessage> keyvalue = iterator.next();
if (CommonUtil.isRecordExpired(new EffectiveDatedAssistantMessage(keyvalue.value))) {
store.delete(keyvalue.key);
}
}
}
LOGGER.info("Finished execution of task for deletion of assistant group messages.");
});
}
其他处理器的init方法中的代码几乎一样。 我正在使用默认的 RocksDB 配置。任何帮助将不胜感激。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。