启动一个 Stream 应用程序(使用 Kafka Streams)失败，并显示"java.lang.IllegalStateException：这不应该发生，因为 headers() 应该只在处理记录时调用">
这似乎只发生在我启动应用程序时，如果主题中已经有数据。如果主题为空并且我开始向它推送数据，则一切都很好。
有人知道为什么会这样吗？
谢谢
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)
基于Matthias分享的信息，这里有一个实际的例子：
- 返回相同的转换器对象(也对处理器和价值转换器有效)会导致
IllegalStateException：
var transformer = new Transformer();
inputStream
.transformValues(() -> transformer)
// further operations...
- 每次返回一个新实例可防止异常：
inputStream
.transformValues(() -> new Transformer())
// further operations...
p.s：我知道文档很清楚，但希望这样可以节省一些滚动。