You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2025/12/11 05:14:22 UTC
(kafka) branch 3.9 updated: KAFKA-19930 Updating doc for process exceptional handler in case of global threads (#21109)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new cc14289cf7c KAFKA-19930 Updating doc for process exceptional handler in case of global threads (#21109)
cc14289cf7c is described below
commit cc14289cf7c7c7b45c5ac3da2d3172f02d263a30
Author: Arpit Goyal <go...@gmail.com>
AuthorDate: Thu Dec 11 10:12:26 2025 +0530
KAFKA-19930 Updating doc for process exceptional handler in case of global threads (#21109)
Cherrypicking from this https://github.com/apache/kafka/pull/21016
Reviewers: Matthias J. Sax <ma...@confluent.io>
---
docs/streams/developer-guide/config-streams.html | 5 ++++-
.../java/org/apache/kafka/streams/StreamsConfig.java | 4 +++-
.../streams/processor/internals/ProcessorNode.java | 9 +++++++++
.../streams/processor/internals/ProcessorNodeTest.java | 18 ++++++++++++++++++
4 files changed, 34 insertions(+), 2 deletions(-)
diff --git a/docs/streams/developer-guide/config-streams.html b/docs/streams/developer-guide/config-streams.html
index 68137dc4b49..7d7ec135826 100644
--- a/docs/streams/developer-guide/config-streams.html
+++ b/docs/streams/developer-guide/config-streams.html
@@ -937,7 +937,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 61cb804b465..ce2aa3c8dfd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
+++ b/streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
@@ -567,7 +567,9 @@ public class StreamsConfig extends AbstractConfig {
/** {@code processing.exception.handler} */
@SuppressWarnings("WeakerAccess")
public static final String PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG = "processing.exception.handler";
- 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 default.dsl.store} */
@Deprecated
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 c1f9c423049..f707fb3b9ea 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
@@ -207,6 +207,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.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 7f4e2d08491..4dafdf04a87 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
@@ -122,6 +122,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",