This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch 4.2
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.2 by this push:
new c36f1cb0e48 KAFKA-19930: Adding process handler null check for global
threads (#21016)
c36f1cb0e48 is described below
commit c36f1cb0e48573082911be6b0d4c9d1b9d09e38b
Author: Arpit Goyal <[email protected]>
AuthorDate: Mon Dec 1 14:37:21 2025 +0530
KAFKA-19930: Adding process handler null check for global threads (#21016)
1. Added process handler null check for global threads.
2. Added unit test to verify it.
3. Updated the documentation.
Reviewers: Matthias J. Sax <[email protected]>, Lucas Brutschy
<[email protected]>
---
docs/streams/developer-guide/config-streams.html | 5 ++++-
.../java/org/apache/kafka/streams/StreamsConfig.java | 7 +++++--
.../streams/processor/internals/ProcessorNode.java | 9 +++++++++
.../streams/processor/internals/ProcessorNodeTest.java | 18 ++++++++++++++++++
4 files changed, 36 insertions(+), 3 deletions(-)
diff --git a/docs/streams/developer-guide/config-streams.html
b/docs/streams/developer-guide/config-streams.html
index 54ea3570e58..b5dc419e8db 100644
--- a/docs/streams/developer-guide/config-streams.html
+++ b/docs/streams/developer-guide/config-streams.html
@@ -1036,7 +1036,10 @@ rack.aware.assignment.tags: zone,cluster |
rack.aware.assignment.tags: zone,cl
<div><p>The processing exception handler allows you to manage
exceptions triggered during the processing of a record. The implemented
exception
handler needs to return a <code>FAIL</code> or
<code>CONTINUE</code> depending on the record and the exception thrown.
Returning
<code>FAIL</code> will signal that Streams should shut down and
<code>CONTINUE</code> will signal that Streams should ignore the issue
- and continue processing. The following library built-in
exception handlers are available:</p>
+ and continue processing.</p>
+ <p><strong>Note:</strong> This handler applies only to regular
stream processing tasks. It does not apply to global state store updates
+ (global threads). Exceptions occurring in global threads will
bubble up to the configured uncaught exception handler.</p>
+ <p>The following library built-in exception handlers are
available:</p>
<ul class="simple">
<li><a class="reference external"
href="/{{version}}/javadoc/org/apache/kafka/streams/errors/LogAndContinueProcessingExceptionHandler.html">LogAndContinueProcessingExceptionHandler</a>:
This handler logs the processing exception and then signals
the processing pipeline to continue processing more records.
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
index 1e04c1fd418..402f6f37d2c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -574,7 +574,8 @@ public class StreamsConfig extends AbstractConfig {
public static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG =
"errors.dead.letter.queue.topic.name";
private static final String ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_DOC = "If
not null, the default exception handler will build and send a Dead Letter Queue
record to the topic with the provided name if an error occurs.\n" +
- "If a custom deserialization/production or processing exception
handler is set, this parameter is ignored for this handler.";
+ "If a custom deserialization/production or processing exception
handler is set, this parameter is ignored for this handler.\n" +
+ "Note: This configuration applies only to regular stream
processing tasks. It does not apply to global state store updates (global
threads).";
/** {@code log.summary.interval.ms} */
public static final String LOG_SUMMARY_INTERVAL_MS_CONFIG =
"log.summary.interval.ms";
@@ -652,7 +653,9 @@ public class StreamsConfig extends AbstractConfig {
@SuppressWarnings("WeakerAccess")
public static final String PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG =
"processing.exception.handler";
@Deprecated
- public static final String PROCESSING_EXCEPTION_HANDLER_CLASS_DOC =
"Exception handling class that implements the
<code>org.apache.kafka.streams.errors.ProcessingExceptionHandler</code>
interface.";
+ public static final String PROCESSING_EXCEPTION_HANDLER_CLASS_DOC =
"Exception handling class that implements the
<code>org.apache.kafka.streams.errors.ProcessingExceptionHandler</code>
interface. " +
+ "Note: This handler applies only to regular stream processing
tasks. It does not apply to global state store updates (global threads). " +
+ "Exceptions occurring in global threads will bubble up to the
configured uncaught exception handler.";
/** {@code processing.guarantee} */
@SuppressWarnings("WeakerAccess")
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
index bbf82ff9033..f121c1626ea 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java
@@ -208,6 +208,15 @@ public class ProcessorNode<KIn, VIn, KOut, VOut> {
// while Java distinguishes checked vs unchecked exceptions, other
languages
// like Scala or Kotlin do not, and thus we need to catch
`Exception`
// (instead of `RuntimeException`) to work well with those
languages
+
+ // If the processing exception handler is not set (e.g., for
global threads),
+ // rethrow the exception to let it bubble up to the uncaught
exception handler.
+ // The processing exception handler is only set for regular stream
tasks, not for
+ // global state update tasks which use a different error handling
mechanism.
+ if (processingExceptionHandler == null) {
+ throw processingException;
+ }
+
final ErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
null, // only required to pass for
DeserializationExceptionHandler
internalProcessorContext.recordContext().topic(),
diff --git
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
index 565d6dcf8f9..21e670ef2cd 100644
---
a/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorNodeTest.java
@@ -133,6 +133,24 @@ public class ProcessorNodeTest {
assertDoesNotThrow(() -> node.process(new Record<>(KEY, VALUE,
TIMESTAMP)));
}
+ @Test
+ public void shouldRethrowExceptionWhenProcessingExceptionHandlerIsNull() {
+ // This simulates the global thread case where no
ProcessingExceptionHandler is set
+ final ProcessorNode<Object, Object, Object, Object> node =
+ new ProcessorNode<>(NAME, new
IgnoredInternalExceptionsProcessor(), Collections.emptySet());
+
+ final InternalProcessorContext<Object, Object>
internalProcessorContext = mockInternalProcessorContext();
+ // Initialize without a ProcessingExceptionHandler (simulates global
thread initialization)
+ node.init(internalProcessorContext);
+
+ // The exception should be rethrown since there's no handler to
process it
+ final RuntimeException exception = assertThrows(RuntimeException.class,
+ () -> node.process(new Record<>(KEY, VALUE, TIMESTAMP)));
+
+ assertEquals("Processing exception should be caught and handled by the
processing exception handler.",
+ exception.getMessage());
+ }
+
@ParameterizedTest
@CsvSource({
"FailedProcessingException,java.lang.RuntimeException,Fail processing",
