mjsax commented on code in PR #16093:
URL: https://github.com/apache/kafka/pull/16093#discussion_r1687234808
##########
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:
Should we log here? How do we do this elsewhere? I know that the default
handlers log already (so might be redundant), but a custom handler might not
log...? Just curious.
--
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]
