Bruno Cadonna created KAFKA-17850:
-------------------------------------
Summary: Stop leaking internal exception to
StreamsUncaughtExceptionHandler
Key: KAFKA-17850
URL: https://issues.apache.org/jira/browse/KAFKA-17850
Project: Kafka
Issue Type: Bug
Components: streams
Affects Versions: 3.9.0
Reporter: Bruno Cadonna
In a log file of a Streams application I found the following log line:
{code:java}
[2024-10-08 16:45:03,846] ERROR [i-0ec21c43346d0de25-StreamThread-2]
stream-client [i-0ec21c43346d0de25] Replacing thread in the streams uncaught
exception handler (org.apache.kafka.streams.KafkaStreams)
org.apache.kafka.streams.errors.internals.FailedProcessingException
at
org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:237)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:291)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:270)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:229)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:216)
at
org.apache.kafka.streams.kstream.internals.KStreamMapValues$KStreamMapProcessor.process(KStreamMapValues.java:42)
at
org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:180)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forwardInternal(ProcessorContextImpl.java:291)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:270)
at
org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:229)
at
org.apache.kafka.streams.kstream.internals.TimestampedCacheFlushListener.apply(TimestampedCacheFlushListener.java:45)
at
org.apache.kafka.streams.state.internals.MeteredWindowStore.lambda$setFlushListener$5(MeteredWindowStore.java:163)
at
org.apache.kafka.streams.state.internals.CachingWindowStore.putAndMaybeForward(CachingWindowStore.java:113)
at
org.apache.kafka.streams.state.internals.CachingWindowStore.lambda$init$0(CachingWindowStore.java:86)
at
org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:159)
at
org.apache.kafka.streams.state.internals.NamedCache.flush(NamedCache.java:117)
at
org.apache.kafka.streams.state.internals.ThreadCache.flush(ThreadCache.java:148)
at
org.apache.kafka.streams.state.internals.CachingWindowStore.flushCache(CachingWindowStore.java:414)
at
org.apache.kafka.streams.state.internals.WrappedStateStore.flushCache(WrappedStateStore.java:79)
at
org.apache.kafka.streams.processor.internals.ProcessorStateManager.flushCache(ProcessorStateManager.java:537)
at
org.apache.kafka.streams.processor.internals.StreamTask.flush(StreamTask.java:423)
at
org.apache.kafka.streams.processor.internals.StreamTask.prepareCommit(StreamTask.java:447)
at
org.apache.kafka.streams.processor.internals.TaskExecutor.commitTasksAndMaybeUpdateCommittableOffsets(TaskExecutor.java:147)
at
org.apache.kafka.streams.processor.internals.TaskManager.commitTasksAndMaybeUpdateCommittableOffsets(TaskManager.java:1925)
at
org.apache.kafka.streams.processor.internals.TaskManager.commit(TaskManager.java:1892)
at
org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:1382)
at
org.apache.kafka.streams.processor.internals.StreamThread.runOnceWithoutProcessingThreads(StreamThread.java:1034)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:712)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:672)
Caused by:
org.apache.kafka.streams.StreamsSoakTest$ThrowableValueProcessorSupplier$ThreadReplaceableException
at
org.apache.kafka.streams.StreamsSoakTest$ThrowableValueProcessorSupplier$KaboomableValueProcessor.process(StreamsSoakTest.java:618)
at
org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:180)
... 28 more
{code}
That indicates that the internal exception {{FailedProcessingException}} is
passed to the Streams-specific uncaught exception handler. Streams should not
pass internal exceptions to a public API that is supposed to be implemented by
the user since that means the user needs to be aware of an internal exception.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
