NPE Kafka Streams в MeteredKeyValueStore - PullRequest
Новая
       4

NPE Kafka Streams в MeteredKeyValueStore

0 голосов
/ 09 января 2019

Я пытаюсь запустить очень простой поток, используя ProcessorAPI в Scala. 

class KafkaProcessor extends Processor[String, GenericRecord] {
    private var kvStore: KeyValueStore[String, GenericRecord] = _

    override def init(processorContext: ProcessorContext): Unit = {
        this.kvStore = Stores
            .keyValueStoreBuilder(
                Stores.persistentKeyValueStore("random-mame"),
                Serdes.String,
                new GenericAvroSerde
            )
    }

    override def process(
         key: String,
         value: GenericRecord
    ): Unit = {
        val currentState = Option(kvStore.get(key)) // NPE
        ...
    }
}

Кажется, какой-то внутренний NPE выбрасывается из журналов ошибок: 

Exception in thread "test-4294024b-1390-4c2f-ba8e-e520cca728ff-StreamThread-1" java.lang.NullPointerException
at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.get(MeteredKeyValueStore.java:134)
at writeside.kafka.AggregateKafkaProcessor.process(KafkaProcessor.scala:64)
at writeside.kafka.AggregateKafkaProcessor.process(KafkaProcessor.scala:35)
at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:115)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:146)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:129)
at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:93)
at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84)
at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:351)
at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:104)
at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:413)
at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:862)
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:777)
at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:747)

Это относится к getTime внутри MeteredKeyValueStore. Я не уверен, как это происходит и как я могу предотвратить это.

1 Ответ

0 голосов
/ 09 января 2019

Если вы хотите использовать хранилище, вам нужно объявить хранилище за пределами процессора (т.е. добавить хранилище в StreamBuilder), подключить хранилище (через StreamsBuilder) к процессор.

Внутри процессора вы используете ProcessorContext, чтобы получить ручку в магазине.

Подробнее см. В документах: https://kafka.apache.org/21/documentation/streams/developer-guide/processor-api.html

...