KAFKA-16448: Unify error-callback exception handling

streams/src/main/java/org/apache/kafka/streams/errors/DeserializationExceptionHandler.java

@@ -37,7 +37,7 @@ public interface DeserializationExceptionHandler extends Configurable {
      * @param context processor context
      * @param record record that failed deserialization
      * @param exception the actual exception
-     * @deprecated Since 3.9. Use Please {@link #handle(ErrorHandlerContext, ConsumerRecord, Exception)}
+     * @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext, ConsumerRecord, Exception)} instead.
      */
     @Deprecated
     default DeserializationHandlerResponse handle(final ProcessorContext context,

streams/src/main/java/org/apache/kafka/streams/errors/ProductionExceptionHandler.java

@@ -59,7 +59,7 @@ default ProductionExceptionHandlerResponse handle(final ErrorHandlerContext cont
      *
      * @param record        the record that failed to serialize
      * @param exception     the exception that occurred during serialization
-     * @deprecated Since 3.9. Use {@link #handle(ErrorHandlerContext, ProducerRecord, Exception)} instead.
+     * @deprecated Since 3.9. Use {@link #handleSerializationException(ErrorHandlerContext, ProducerRecord, Exception, SerializationExceptionOrigin)} instead.
      */
     @Deprecated
     default ProductionExceptionHandlerResponse handleSerializationException(final ProducerRecord record,

streams/src/main/java/org/apache/kafka/streams/errors/internals/DefaultErrorHandlerContext.java

@@ -81,6 +81,18 @@ public TaskId taskId() {
         return taskId;
     }
 
+    @Override
+    public String toString() {
+        // we do exclude headers on purpose, to not accidentally log user data
+        return "ErrorHandlerContext{" +
+            "topic='" + topic + '\'' +
+            ", partition=" + partition +
+            ", offset=" + offset +
+            ", processorNodeId='" + processorNodeId + '\'' +
+            ", taskId=" + taskId +
+            '}';
+    }
+
     public Optional<ProcessorContext> processorContext() {
         return Optional.ofNullable(processorContext);
     }

streams/src/main/java/org/apache/kafka/streams/errors/internals/FailedProcessingException.java

@@ -25,7 +25,12 @@
 public class FailedProcessingException extends StreamsException {
     private static final long serialVersionUID = 1L;
 
+    public FailedProcessingException(final String errorMessage, final Exception exception) {
+        super(errorMessage, exception);
+    }
+
     public FailedProcessingException(final Exception exception) {
-        super(exception);
+        // we need to explicitly set `message` to `null` here
+        super(null, exception);
     }
 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java

@@ -319,7 +319,7 @@ private void reprocessState(final List<TopicPartition> topicPartitions,
                                 record.headers()));
                             restoreCount++;
                         }
-                    } catch (final Exception deserializationException) {
+                    } catch (final RuntimeException deserializationException) {
                         handleDeserializationFailure(
                             deserializationExceptionHandler,
                             globalProcessorContext,
@@ -330,7 +330,8 @@ private void reprocessState(final List<TopicPartition> topicPartitions,
                                 Thread.currentThread().getName(),
                                 globalProcessorContext.taskId().toString(),
                                 globalProcessorContext.metrics()
-                            )
+                            ),
+                            null
                         );
                     }
                 }

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorNode.java

@@ -38,6 +38,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 
 import static org.apache.kafka.streams.StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG;
@@ -202,7 +203,7 @@ public void process(final Record<KIn, VIn> record) {
         } catch (final FailedProcessingException | TaskCorruptedException | TaskMigratedException e) {
             // Rethrow exceptions that should not be handled here
             throw e;
-        } catch (final RuntimeException e) {
+        } catch (final RuntimeException processingException) {
             final ErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
                 null, // only required to pass for DeserializationExceptionHandler
                 internalProcessorContext.topic(),
@@ -213,18 +214,26 @@ public void process(final Record<KIn, VIn> record) {
                 internalProcessorContext.taskId());
 
             final ProcessingExceptionHandler.ProcessingHandlerResponse response;
-
             try {
-                response = processingExceptionHandler.handle(errorHandlerContext, record, e);
-            } catch (final Exception fatalUserException) {
-                throw new FailedProcessingException(fatalUserException);
+                response = Objects.requireNonNull(
+                    processingExceptionHandler.handle(errorHandlerContext, record, processingException),
+                    "Invalid ProductionExceptionHandler response."
+                );
+            } catch (final RuntimeException fatalUserException) {
+                log.error(
+                    "Processing error callback failed after processing error for record: {}",
+                    errorHandlerContext,
+                    processingException
+                );
+                throw new FailedProcessingException("Fatal user code error in processing error callback", fatalUserException);
             }
+
             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);
+                throw new FailedProcessingException(processingException);
             } else {
                 droppedRecordsSensor.record();
             }

streams/src/main/java/org/apache/kafka/streams/processor/internals/RecordCollectorImpl.java

@@ -45,6 +45,7 @@
 import org.apache.kafka.streams.errors.TaskCorruptedException;
 import org.apache.kafka.streams.errors.TaskMigratedException;
 import org.apache.kafka.streams.errors.internals.DefaultErrorHandlerContext;
+import org.apache.kafka.streams.errors.internals.FailedProcessingException;
 import org.apache.kafka.streams.processor.StreamPartitioner;
 import org.apache.kafka.streams.processor.TaskId;
 import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
@@ -59,6 +60,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
@@ -208,7 +210,7 @@ public <K, V> void send(final String topic,
                 key,
                 keySerializer,
                 exception);
-        } catch (final Exception exception) {
+        } catch (final RuntimeException serializationException) {
             handleException(
                 ProductionExceptionHandler.SerializationExceptionOrigin.KEY,
                 topic,
@@ -219,7 +221,7 @@ public <K, V> void send(final String topic,
                 timestamp,
                 processorNodeId,
                 context,
-                exception);
+                serializationException);
             return;
         }
 
@@ -232,7 +234,7 @@ public <K, V> void send(final String topic,
                 value,
                 valueSerializer,
                 exception);
-        } catch (final Exception exception) {
+        } catch (final RuntimeException serializationException) {
             handleException(
                 ProductionExceptionHandler.SerializationExceptionOrigin.VALUE,
                 topic,
@@ -243,7 +245,7 @@ public <K, V> void send(final String topic,
                 timestamp,
                 processorNodeId,
                 context,
-                exception);
+                serializationException);
             return;
         }
 
@@ -297,42 +299,51 @@ private <K, V> void handleException(final ProductionExceptionHandler.Serializati
                                         final Long timestamp,
                                         final String processorNodeId,
                                         final InternalProcessorContext<Void, Void> context,
-                                        final Exception exception) {
+                                        final RuntimeException serializationException) {
+        log.debug(String.format("Error serializing record for topic %s", topic), serializationException);
+
+        final DefaultErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
+            null, // only required to pass for DeserializationExceptionHandler
+            context.recordContext().topic(),
+            context.recordContext().partition(),
+            context.recordContext().offset(),
+            context.recordContext().headers(),
+            processorNodeId,
+            taskId
+        );
         final ProducerRecord<K, V> record = new ProducerRecord<>(topic, partition, timestamp, key, value, headers);
-        final ProductionExceptionHandlerResponse response;
-
-        log.debug(String.format("Error serializing record to topic %s", topic), exception);
 
+        final ProductionExceptionHandlerResponse response;
         try {
-            final DefaultErrorHandlerContext errorHandlerContext = new DefaultErrorHandlerContext(
-                null, // only required to pass for DeserializationExceptionHandler
-                context.recordContext().topic(),
-                context.recordContext().partition(),
-                context.recordContext().offset(),
-                context.recordContext().headers(),
-                processorNodeId,
-                taskId
+            response = Objects.requireNonNull(
+                productionExceptionHandler.handleSerializationException(errorHandlerContext, record, serializationException, origin),
+                "Invalid ProductionExceptionHandler response."
             );
-            response = productionExceptionHandler.handleSerializationException(errorHandlerContext, record, exception, origin);
-        } catch (final Exception e) {
-            log.error("Fatal when handling serialization exception", e);
-            recordSendError(topic, e, null, context, processorNodeId);
-            return;
+        } catch (final RuntimeException fatalUserException) {
+            log.error(
+                String.format(
+                    "Production error callback failed after serialization error for record %s: %s",
+                    origin.toString().toLowerCase(Locale.ROOT),
+                    errorHandlerContext
+                ),
+                serializationException
+            );
+            throw new FailedProcessingException("Fatal user code error in production error callback", fatalUserException);
         }
 
         if (response == ProductionExceptionHandlerResponse.FAIL) {
             throw new StreamsException(
                 String.format(
                     "Unable to serialize record. ProducerRecord(topic=[%s], partition=[%d], timestamp=[%d]",
-                        topic,
-                        partition,
-                        timestamp),
-                    exception
+                    topic,
+                    partition,
+                    timestamp),
+                serializationException
             );
         }
 
         log.warn("Unable to serialize record, continue processing. " +
-                        "ProducerRecord(topic=[{}], partition=[{}], timestamp=[{}])",
+                    "ProducerRecord(topic=[{}], partition=[{}], timestamp=[{}])",
                 topic,
                 partition,
                 timestamp);
@@ -364,24 +375,24 @@ private <KV> StreamsException createStreamsExceptionForClassCastException(final
     }
 
     private void recordSendError(final String topic,
-                                 final Exception exception,
+                                 final Exception productionException,
                                  final ProducerRecord<byte[], byte[]> serializedRecord,
                                  final InternalProcessorContext<Void, Void> context,
                                  final String processorNodeId) {
-        String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic, taskId, exception.toString());
+        String errorMessage = String.format(SEND_EXCEPTION_MESSAGE, topic, taskId, productionException.toString());
 
-        if (isFatalException(exception)) {
+        if (isFatalException(productionException)) {
             errorMessage += "\nWritten offsets would not be recorded and no more records would be sent since this is a fatal error.";
-            sendException.set(new StreamsException(errorMessage, exception));
-        } else if (exception instanceof ProducerFencedException ||
-                exception instanceof InvalidPidMappingException ||
-                exception instanceof InvalidProducerEpochException ||
-                exception instanceof OutOfOrderSequenceException) {
+            sendException.set(new StreamsException(errorMessage, productionException));
+        } else if (productionException instanceof ProducerFencedException ||
+                productionException instanceof InvalidPidMappingException ||
+                productionException instanceof InvalidProducerEpochException ||
+                productionException instanceof OutOfOrderSequenceException) {
             errorMessage += "\nWritten offsets would not be recorded and no more records would be sent since the producer is fenced, " +
                 "indicating the task may be migrated out";
-            sendException.set(new TaskMigratedException(errorMessage, exception));
+            sendException.set(new TaskMigratedException(errorMessage, productionException));
         } else {
-            if (isRetriable(exception)) {
+            if (isRetriable(productionException)) {
                 errorMessage += "\nThe broker is either slow or in bad state (like not having enough replicas) in responding the request, " +
                     "or the connection to broker was interrupted sending the request or receiving the response. " +
                     "\nConsider overwriting `max.block.ms` and /or " +
@@ -398,17 +409,34 @@ private void recordSendError(final String topic,
                     taskId
                 );
 
-                if (productionExceptionHandler.handle(errorHandlerContext, serializedRecord, exception) == ProductionExceptionHandlerResponse.FAIL) {
+                final ProductionExceptionHandlerResponse response;
+                try {
+                    response = Objects.requireNonNull(
+                        productionExceptionHandler.handle(errorHandlerContext, serializedRecord, productionException),
+                        "Invalid ProductionExceptionHandler response."
+                    );
+                } catch (final RuntimeException fatalUserException) {
+                    log.error(
+                        "Production error callback failed after production error for record {}",
+                        serializedRecord,
+                        productionException
+                    );
+                    sendException.set(new FailedProcessingException("Fatal user code error in production error callback", fatalUserException));
+                    return;
+                }
+
+                if (response == ProductionExceptionHandlerResponse.FAIL) {
                     errorMessage += "\nException handler choose to FAIL the processing, no more records would be sent.";
-                    sendException.set(new StreamsException(errorMessage, exception));
+                    sendException.set(new StreamsException(errorMessage, productionException));
                 } else {
                     errorMessage += "\nException handler choose to CONTINUE processing in spite of this error but written offsets would not be recorded.";
                     droppedRecordsSensor.record();
                 }
+
             }
         }
 
-        log.error(errorMessage, exception);
+        log.error(errorMessage, productionException);
     }
 
     /**