[ https://issues.apache.org/jira/browse/KAFKA-15463?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17779715#comment-17779715 ]
A. Sophie Blee-Goldman commented on KAFKA-15463: ------------------------------------------------ [~yevsh] Well the intended pattern is to get the StateStore from the context during init and then save a reference to the store for your processor/transformer. That's why we pass the context in to #init but not to #process (also, while it's not a significant overhead, it's definitely more efficient to do the state store lookup only once, during init, and not on every single record that's processed) That said, I think it's a fair question as to why this is causing an error, and why the error doesn't happen every time. The fact that it generally only appears when you have more than one partition/task per application instance makes me think there is some state that is somehow being shared between the different tasks. This would definitely be explained by returning the same transformer instance each time, but based on your latest update, you are definitely returning a new transformer each time and still seeing the issue, right? Assuming so, I'm inclined to believe the issue is with Spring. I vaguely recall a similar problem being reported in the past, which was ultimately because a Spring object was unexpectedly/unknowingly acting as a static/singleton class. This was resulting in the context – of which there is supposed to be exactly one per task – being shared between different instances of the task/processor nodes. I'm pretty sure that's what is going on here as well. I'm guessing that the MyService class is a Spring bean? If so, then it's effectively a singleton and will be shared by each of the individual Transformer instances in your application, meaning the different tasks will be overwriting each others context when invoking #init on this transformer. So the context you retrieve from the myServiceBean during #process may not be the same as the context you saved to it in #init, causing it to throw this error since only the context corresponding to the task that is currently being processed will have the currentNode set to a non-null value. Even if you made the change I originally suggested but saved the StateStore reference by passing it to the MyService bean, it wouldn't work – it might not throw this error but you would potentially be reading and writing to the wrong copy of a state store for a given task, which is even worse. The only way to solve this is by removing the Spring bean entirely or at least refactoring it so that it doesn't hold any internal state and has to have the full application state for that task passed in to it every time – in other words you just need to make sure to keep all the objects used by a given transformer completely local to that instance. Here is my recommendation for how to implement your transformer class – hope this helps! private final MYService myServiceBean; private StateStore myStore; @Overridepublic void init(ProcessorContext context) \{ myStore = context.getStateStore(STORE_NAME); } @Overridepublic KeyValue<String, MyItem> transform(String key, MyItem myItem) \{ myServiceBean.process(myItem, myStore); } Basically modify the MyService bean to accept the StateStore to operate on as a parameter to its #process method. And definitely keep the fix in which you return a new Transformer instance each time instead of reusing the same one. Let me know if you have any questions! I'm going to close the ticket since I'm fairly confident in this solution having seen the same problem before, but definitely please do reopen it if you implement the suggested fix and still encounter an error. Good luck! > StreamsException: Accessing from an unknown node > ------------------------------------------------- > > Key: KAFKA-15463 > URL: https://issues.apache.org/jira/browse/KAFKA-15463 > Project: Kafka > Issue Type: Bug > Components: streams > Affects Versions: 3.2.1 > Reporter: Yevgeny > Priority: Major > > After some time application was working fine, starting to get: > > This is springboot application runs in kubernetes as stateful pod. > > > > {code:java} > Exception in thread > "xxxxxxxxxxxx-ddf9819f-d6c7-46ce-930e-cd923e1b3c2c-StreamThread-1" > org.apache.kafka.streams.errors.StreamsException: Accessing from an unknown > node at > org.apache.kafka.streams.processor.internals.ProcessorContextImpl.getStateStore(ProcessorContextImpl.java:162) > at myclass1.java:28) at myclass2.java:48) at > java.base/java.util.stream.MatchOps$1MatchSink.accept(MatchOps.java:90) at > java.base/java.util.ArrayList$ArrayListSpliterator.tryAdvance(ArrayList.java:1602) > at > java.base/java.util.stream.ReferencePipeline.forEachWithCancel(ReferencePipeline.java:129) > at > java.base/java.util.stream.AbstractPipeline.copyIntoWithCancel(AbstractPipeline.java:527) > at > java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:513) > at > java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) > at > java.base/java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:230) > at > java.base/java.util.stream.MatchOps$MatchOp.evaluateSequential(MatchOps.java:196) > at > java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) > at > java.base/java.util.stream.ReferencePipeline.allMatch(ReferencePipeline.java:637) > at myclass3.java:48) at > org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:49) > at > org.apache.kafka.streams.kstream.internals.TransformerSupplierAdapter$1.transform(TransformerSupplierAdapter.java:38) > at > org.apache.kafka.streams.kstream.internals.KStreamFlatTransform$KStreamFlatTransformProcessor.process(KStreamFlatTransform.java:66) > at > org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:146) > at > org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:275) > at > org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:254) > at > org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:213) > at > org.apache.kafka.streams.processor.internals.SourceNode.process(SourceNode.java:84) > at > org.apache.kafka.streams.processor.internals.StreamTask.lambda$doProcess$1(StreamTask.java:780) > at > org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.maybeMeasureLatency(StreamsMetricsImpl.java:809) > at > org.apache.kafka.streams.processor.internals.StreamTask.doProcess(StreamTask.java:780) > at > org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:711) > at > org.apache.kafka.streams.processor.internals.TaskExecutor.processTask(TaskExecutor.java:100) > at > org.apache.kafka.streams.processor.internals.TaskExecutor.process(TaskExecutor.java:81) > at > org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1177) > at > org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:769) > at > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:589) > at > org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:551) > {code} > > stream-thread > [xxxxxxxxxxxx-ddf9819f-d6c7-46ce-930e-cd923e1b3c2c-StreamThread-1] State > transition from PENDING_SHUTDOWN to DEAD > > > Transformer is Prototype bean, the supplier supplys new instance of the > Transformer: > > > {code:java} > @Override public Transformer<String, MyItem, KeyValue<String, MyItem>> get() > { return ctx.getBean(MyTransformer.class); }{code} > > > The only way to recover is to delete all topics used by kafkastreams, even if > application restarted same exception is thrown. > *If messages in internal topics of 'store-changelog' are deleted/offset > manipulated, can it cause the issue? -- This message was sent by Atlassian Jira (v8.20.10#820010)