cadonna commented on code in PR #16093:
URL: https://github.com/apache/kafka/pull/16093#discussion_r1687514719
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java:
##########
@@ -174,6 +200,32 @@ public void process(final Record<KIn, VIn> record) {
keyClass,
valueClass),
e);
+ } catch (final FailedProcessingException | TaskCorruptedException |
TaskMigratedException e) {
+ // Rethrow exceptions that should not be handled here
+ throw e;
+ } catch (final Exception e) {
+ final ErrorHandlerContext errorHandlerContext = new
DefaultErrorHandlerContext(
+ internalProcessorContext.topic(),
+ internalProcessorContext.partition(),
+ internalProcessorContext.offset(),
+ internalProcessorContext.headers(),
+ internalProcessorContext.recordContext().rawRecord().key(),
+ internalProcessorContext.recordContext().rawRecord().value(),
+ internalProcessorContext.currentNode().name(),
+ internalProcessorContext.taskId());
+
+ final ProcessingExceptionHandler.ProcessingHandlerResponse
response = processingExceptionHandler
+ .handle(errorHandlerContext, record, e);
+
+ if (response ==
ProcessingExceptionHandler.ProcessingHandlerResponse.FAIL) {
+ log.error("Processing exception handler is set to fail upon" +
+ " a processing error. If you would rather have the
streaming pipeline" +
+ " continue after a processing error, please set the " +
+ PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG + "
appropriately.");
+ throw new FailedProcessingException(e);
+ } else {
+ droppedRecordsSensor.record();
Review Comment:
I also thought about logging here. But since we had quite some issues with
log pressure due to dropped records, I thought, it would be better to let the
handler log.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
