java.lang.IllegalStateException: это не должно происходить, поскольку headers () должен вызываться только во время обработки записи - PullRequest
Новая
       2

java.lang.IllegalStateException: это не должно происходить, поскольку headers () должен вызываться только во время обработки записи

0 голосов
/ 29 мая 2019

Запуск приложения Stream (с использованием Kafka Streams) завершается с ошибкой «java.lang.IllegalStateException: это не должно происходить, поскольку заголовки () должны вызываться только во время обработки записи»

Похоже, это происходит только тогда, когда я запускаю приложение, если в теме уже есть данные. Если тема пуста, и я начинаю загружать данные в нее, все в порядке.

Кто-нибудь знает, почему это произойдет?

Спасибо 

This should not happen as headers() should only be called while a record is processed
java.lang.IllegalStateException: This should not happen as headers() should only be called while a record is processed
        at org.apache.kafka.streams.processor.internals.AbstractProcessorContext.headers(AbstractProcessorContext.java:149)
        at org.apache.kafka.streams.state.internals.CachingKeyValueStore.putInternal(CachingKeyValueStore.java:235)
        at org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:222)
        at org.apache.kafka.streams.state.internals.CachingKeyValueStore.put(CachingKeyValueStore.java:37)
        at org.apache.kafka.streams.state.internals.MeteredKeyValueStore.put(MeteredKeyValueStore.java:153)
        at org.apache.kafka.streams.processor.internals.ProcessorContextImpl$KeyValueStoreReadWriteDecorator.put(ProcessorContextImpl.java:446)
        at fr.mediametrie.GeoIPEnrichmentProcessor.transform(GeoIPEnrichmentProcessor.java:97)
        at fr.mediametrie.GeoIPEnrichmentProcessor.transform(GeoIPEnrichmentProcessor.java:1)
        at org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:47)
        at org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:36)
        at org.apache.kafka.streams.kstream.internals.KStreamFlatTransform$KStreamFlatTransformProcessor.process(KStreamFlatTransform.java:56)
        at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
        at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:183)
        at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:162)
        at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:122)
        at org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:87)
        at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:364)
        at org.apache.kafka.streams.processor.internals.AssignedStreamsTasks.process(AssignedStreamsTasks.java:199)
        at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:420)
        at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:890)
        at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:805)
        at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:774)
...