如何从kafkastreams persistentkeyvaluestore中删除记录?

3mpgtkmj  于 2021-06-07  发布在  Kafka
关注(0)|答案(1)|浏览(246)

我将一些记录放入persistentkeyvaluestore:

Long nowValue = Optional.ofNullable(myStore.get(key)).orElse(0L) + newValue;
myStore.put(key, nowValue);

它运行良好。我可以通过println查看记录:

KeyValueIterator<String, Long> iter = myStore.all();
while (iter.hasNext()) {
     KeyValue<String, Long> entry = iter.next();
     System.out.println(entry.value)
}
iter.close();

但当我试图删除一个记录时:

KeyValueIterator<String, Long> iter = myStore.all();
while (iter.hasNext()) {
     iter.remove();
}
iter.close();

转到异常:

java.lang.UnsupportedOperationException: RocksDB iterator does not support remove()
    at org.apache.kafka.streams.state.internals.RocksDBStore$RocksDbIterator.remove(RocksDBStore.java:515)
    at org.apache.kafka.streams.state.internals.InnerMeteredKeyValueStore$MeteredKeyValueIterator.remove(InnerMeteredKeyValueStore.java:306)
    at com.loogtech.bi.compute.realtime.processor.AddCashProcessor$TotalCashAddPunctuator.punctuate(AddCashProcessor.java:115)
    at org.apache.kafka.streams.processor.internals.ProcessorNode$4.run(ProcessorNode.java:131)
    at org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:208)
    at org.apache.kafka.streams.processor.internals.ProcessorNode.punctuate(ProcessorNode.java:134)
    at org.apache.kafka.streams.processor.internals.StreamTask.punctuate(StreamTask.java:263)
    at org.apache.kafka.streams.processor.internals.PunctuationQueue.mayPunctuate(PunctuationQueue.java:54)
    at org.apache.kafka.streams.processor.internals.StreamTask.maybePunctuateSystemTime(StreamTask.java:619)
    at org.apache.kafka.streams.processor.internals.AssignedTasks.punctuate(AssignedTasks.java:430)
    at org.apache.kafka.streams.processor.internals.TaskManager.punctuate(TaskManager.java:324)
    at org.apache.kafka.streams.processor.internals.StreamThread.punctuate(StreamThread.java:969)
    at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:834)
    at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:774)
    at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:744)

似乎也不能删除记录 myStore.remove(key) 如何按键删除记录?

eqqqjvef

eqqqjvef1#

接口 KeyValueStore 提供 #delete(...) 方法(非 #remove(...) )从存储中删除记录。

相关问题