This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 09be14bb09d KAFKA-16448: Fix processing exception handler (#16663)
09be14bb09d is described below
commit 09be14bb09dc336f941a7859232094bfb3cb3b96
Author: Sebastien Viale <sebastien.vi...@michelin.com>
AuthorDate: Fri Jul 26 01:17:31 2024 +0200
KAFKA-16448: Fix processing exception handler (#16663)
Co-authored-by: Dabz <d.gaspar...@gmail.com>
Co-authored-by: loicgreffier <loic.greff...@michelin.com>
Minor code improvements across different classed, related to the
`ProcessingExceptionHandler` implementation (KIP-1033).
Reviewers: Bruno Cadonna <br...@confluent.io>, Matthias J. Sax
<matth...@confluent.io>
---
.../LogAndFailProcessingExceptionHandler.java | 2 +-
.../internals/DefaultErrorHandlerContext.java | 16 ++++-----
.../internals/FailedProcessingException.java | 8 ++---
.../streams/processor/internals/StreamTask.java | 42 ++++++++++++----------
4 files changed, 36 insertions(+), 32 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailProcessingExceptionHandler.java
b/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailProcessingExceptionHandler.java
index 47fdb09c9c2..9c2cf91c605 100644
---
a/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailProcessingExceptionHandler.java
+++
b/streams/src/main/java/org/apache/kafka/streams/errors/LogAndFailProcessingExceptionHandler.java
@@ -32,7 +32,7 @@ public class LogAndFailProcessingExceptionHandler implements
ProcessingException
@Override
public ProcessingHandlerResponse handle(final ErrorHandlerContext context,
final Record<?, ?> record, final Exception exception) {
- log.warn("Exception caught during message processing, " +
+ log.error("Exception caught during message processing, " +
"processor node: {}, taskId: {}, source topic: {}, source
partition: {}, source offset: {}",
context.processorNodeId(), context.taskId(), context.topic(),
context.partition(), context.offset(),
exception);
diff --git
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
index fc6b6048cb9..ff79860d77e 100644
---
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
+++
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java
@@ -53,41 +53,41 @@ public class DefaultErrorHandlerContext implements
ErrorHandlerContext {
@Override
public String topic() {
- return this.topic;
+ return topic;
}
@Override
public int partition() {
- return this.partition;
+ return partition;
}
@Override
public long offset() {
- return this.offset;
+ return offset;
}
@Override
public Headers headers() {
- return this.headers;
+ return headers;
}
@Override
public byte[] sourceRawKey() {
- return this.sourceRawKey;
+ return sourceRawKey;
}
@Override
public byte[] sourceRawValue() {
- return this.sourceRawValue;
+ return sourceRawValue;
}
@Override
public String processorNodeId() {
- return this.processorNodeId;
+ return processorNodeId;
}
@Override
public TaskId taskId() {
- return this.taskId;
+ return taskId;
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
index 81b2a2d4fb1..25f2ae9f6cc 100644
---
a/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
+++
b/streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java
@@ -16,16 +16,16 @@
*/
package org.apache.kafka.streams.errors.internals;
-import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.streams.errors.StreamsException;
/**
* {@link FailedProcessingException} is the top-level exception type generated
by Kafka Streams, and indicates errors have
* occurred during a {@link
org.apache.kafka.streams.processor.internals.ProcessorNode ProcessorNode's}
processing.
*/
-public class FailedProcessingException extends KafkaException {
+public class FailedProcessingException extends StreamsException {
private static final long serialVersionUID = 1L;
- public FailedProcessingException(final Throwable throwable) {
- super(throwable);
+ public FailedProcessingException(final Exception exception) {
+ super(exception);
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
index 52d40a34915..30b9038aa6a 100644
---
a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
+++
b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java
@@ -800,28 +800,14 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
record = null;
throw new TaskCorruptedException(Collections.singleton(id));
}
+ } catch (final FailedProcessingException failedProcessingException) {
+ // Do not keep the failed processing exception in the stack trace
+ handleException(failedProcessingException.getCause());
} catch (final StreamsException exception) {
record = null;
throw exception;
- } catch (final RuntimeException e) {
- // Do not keep the failed processing exception in the stack trace
- final Throwable processingException = e instanceof
FailedProcessingException ? e.getCause() : e;
-
- final StreamsException error = new StreamsException(
- String.format(
- "Exception caught in process. taskId=%s, processor=%s,
topic=%s, partition=%d, offset=%d, stacktrace=%s",
- id(),
- processorContext.currentNode().name(),
- record.topic(),
- record.partition(),
- record.offset(),
- getStacktraceString(processingException)
- ),
- processingException
- );
- record = null;
-
- throw error;
+ } catch (final Exception e) {
+ handleException(e);
} finally {
processorContext.setCurrentNode(null);
}
@@ -829,6 +815,24 @@ public class StreamTask extends AbstractTask implements
ProcessorNodePunctuator,
return true;
}
+ private void handleException(final Throwable e) {
+ final StreamsException error = new StreamsException(
+ String.format(
+ "Exception caught in process. taskId=%s, processor=%s,
topic=%s, partition=%d, offset=%d, stacktrace=%s",
+ id(),
+ processorContext.currentNode().name(),
+ record.topic(),
+ record.partition(),
+ record.offset(),
+ getStacktraceString(e)
+ ),
+ e
+ );
+ record = null;
+
+ throw error;
+ }
+
@SuppressWarnings("unchecked")
private void doProcess(final long wallClockTime) {
// process the record by passing to the source node of the topology
