KAFKA-16448: Catch and handle processing exceptions

streams/src/main/java/org/apache/kafka/streams/TopologyConfig.java

@@ -22,6 +22,7 @@
 import org.apache.kafka.common.config.ConfigDef.Type;
 import org.apache.kafka.common.utils.Utils;
 import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
+import org.apache.kafka.streams.errors.ProcessingExceptionHandler;
 import org.apache.kafka.streams.internals.StreamsConfigUtils;
 import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.internals.MaterializedInternal;
@@ -53,6 +54,7 @@
 import static org.apache.kafka.streams.StreamsConfig.IN_MEMORY;
 import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_CONFIG;
 import static org.apache.kafka.streams.StreamsConfig.MAX_TASK_IDLE_MS_DOC;
+import static org.apache.kafka.streams.StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG;
 import static org.apache.kafka.streams.StreamsConfig.ROCKS_DB;
 import static org.apache.kafka.streams.StreamsConfig.STATESTORE_CACHE_MAX_BYTES_CONFIG;
 import static org.apache.kafka.streams.StreamsConfig.STATESTORE_CACHE_MAX_BYTES_DOC;
@@ -136,6 +138,7 @@ public class TopologyConfig extends AbstractConfig {
     public final Class<?> dslStoreSuppliers;
     public final Supplier<TimestampExtractor> timestampExtractorSupplier;
     public final Supplier<DeserializationExceptionHandler> deserializationExceptionHandlerSupplier;
+    public final Supplier<ProcessingExceptionHandler> processingExceptionHandlerSupplier;
 
     public TopologyConfig(final StreamsConfig globalAppConfigs) {
         this(null, globalAppConfigs, new Properties());
@@ -151,6 +154,7 @@ public TopologyConfig(final String topologyName, final StreamsConfig globalAppCo
 
         this.applicationConfigs = globalAppConfigs;
         this.topologyOverrides = topologyOverrides;
+        this.processingExceptionHandlerSupplier = () -> globalAppConfigs.getConfiguredInstance(PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG, ProcessingExceptionHandler.class);
 
         if (isTopologyOverride(BUFFERED_RECORDS_PER_PARTITION_CONFIG, topologyOverrides)) {
             maxBufferedSize = getInt(BUFFERED_RECORDS_PER_PARTITION_CONFIG);
@@ -281,6 +285,7 @@ public TaskConfig getTaskConfig() {
             maxBufferedSize,
             timestampExtractorSupplier.get(),
             deserializationExceptionHandlerSupplier.get(),
+            processingExceptionHandlerSupplier.get(),
             eosEnabled
         );
     }
@@ -291,19 +296,22 @@ public static class TaskConfig {
         public final int maxBufferedSize;
         public final TimestampExtractor timestampExtractor;
         public final DeserializationExceptionHandler deserializationExceptionHandler;
+        public final ProcessingExceptionHandler processingExceptionHandler;
         public final boolean eosEnabled;
 
         private TaskConfig(final long maxTaskIdleMs,
                            final long taskTimeoutMs,
                            final int maxBufferedSize,
                            final TimestampExtractor timestampExtractor,
                            final DeserializationExceptionHandler deserializationExceptionHandler,
+                           final ProcessingExceptionHandler processingExceptionHandler,
                            final boolean eosEnabled) {
             this.maxTaskIdleMs = maxTaskIdleMs;
             this.taskTimeoutMs = taskTimeoutMs;
             this.maxBufferedSize = maxBufferedSize;
             this.timestampExtractor = timestampExtractor;
             this.deserializationExceptionHandler = deserializationExceptionHandler;
+            this.processingExceptionHandler = processingExceptionHandler;
             this.eosEnabled = eosEnabled;
         }
     }

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

@@ -20,6 +20,9 @@
 import org.apache.kafka.streams.errors.ErrorHandlerContext;
 import org.apache.kafka.streams.processor.TaskId;
 
+/**
+ * Default implementation of {@link ErrorHandlerContext} that provides access to the metadata of the record that caused the error.
+ */
 public class DefaultErrorHandlerContext implements ErrorHandlerContext {
     private final String topic;
     private final int partition;

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

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.errors.internals;
+
+import org.apache.kafka.common.KafkaException;
+
+/**
+ * {@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 {
+    private static final long serialVersionUID = 1L;
+
+    public FailedProcessingException(final Throwable throwable) {
+        super(throwable);
+    }
+}

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

@@ -28,7 +28,7 @@
 public class CorruptedRecord extends StampedRecord {
 
     CorruptedRecord(final ConsumerRecord<byte[], byte[]> rawRecord) {
-        super(rawRecord, ConsumerRecord.NO_TIMESTAMP);
+        super(rawRecord, ConsumerRecord.NO_TIMESTAMP, rawRecord);
     }
 
     @Override

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

@@ -307,7 +307,8 @@ private void reprocessState(final List<TopicPartition> topicPartitions,
                             record.offset(),
                             record.partition(),
                             record.topic(),
-                            record.headers());
+                            record.headers(),
+                            record);
                     globalProcessorContext.setRecordContext(recordContext);
 
                     try {

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

@@ -113,7 +113,8 @@ public void update(final ConsumerRecord<byte[], byte[]> record) {
                     deserialized.offset(),
                     deserialized.partition(),
                     deserialized.topic(),
-                    deserialized.headers());
+                    deserialized.headers(),
+                    record);
             processorContext.setRecordContext(recordContext);
             processorContext.setCurrentNode(sourceNodeAndDeserializer.sourceNode());
             final Record<Object, Object> toProcess = new Record<>(

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

@@ -66,7 +66,8 @@ public void process(final Record<KIn, VIn> record) {
                 context.offset(),
                 context.partition(),
                 context.topic(),
-                record.headers()
+                record.headers(),
+                processorRecordContext.rawRecord()
             ));
             delegate.process(record.key(), record.value());
         } finally {

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

@@ -261,7 +261,8 @@ public <K, V> void forward(final Record<K, V> record, final String childName) {
                     recordContext.offset(),
                     recordContext.partition(),
                     recordContext.topic(),
-                    record.headers());
+                    record.headers(),
+                    recordContext.rawRecord());
             }
 
             if (childName == null) {

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

@@ -16,22 +16,36 @@
  */
 package org.apache.kafka.streams.processor.internals;
 
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.streams.errors.ErrorHandlerContext;
+import org.apache.kafka.streams.errors.ProcessingExceptionHandler;
 import org.apache.kafka.streams.errors.StreamsException;
+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.Punctuator;
 import org.apache.kafka.streams.processor.api.FixedKeyProcessor;
 import org.apache.kafka.streams.processor.api.FixedKeyProcessorContext;
 import org.apache.kafka.streams.processor.api.InternalFixedKeyRecordFactory;
 import org.apache.kafka.streams.processor.api.Processor;
 import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.processor.internals.metrics.TaskMetrics;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 
+import static org.apache.kafka.streams.StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG;
+
 public class ProcessorNode<KIn, VIn, KOut, VOut> {
 
+    private final Logger log = LoggerFactory.getLogger(ProcessorNode.class);
     private final List<ProcessorNode<KOut, VOut, ?, ?>> children;
     private final Map<String, ProcessorNode<KOut, VOut, ?, ?>> childByName;
 
@@ -40,12 +54,15 @@ public class ProcessorNode<KIn, VIn, KOut, VOut> {
     private final String name;
 
     public final Set<String> stateStores;
+    private ProcessingExceptionHandler processingExceptionHandler;
 
     private InternalProcessorContext<KOut, VOut> internalProcessorContext;
     private String threadId;
 
     private boolean closed = true;
 
+    private Sensor droppedRecordsSensor;
+
     public ProcessorNode(final String name) {
         this(name, (Processor<KIn, VIn, KOut, VOut>) null, null);
     }
@@ -98,6 +115,10 @@ public void init(final InternalProcessorContext<KOut, VOut> context) {
         try {
             threadId = Thread.currentThread().getName();
             internalProcessorContext = context;
+            droppedRecordsSensor = TaskMetrics.droppedRecordsSensor(threadId,
+                internalProcessorContext.taskId().toString(),
+                internalProcessorContext.metrics());
+
             if (processor != null) {
                 processor.init(context);
             }
@@ -115,6 +136,11 @@ public void init(final InternalProcessorContext<KOut, VOut> context) {
         closed = false;
     }
 
+    public void init(final InternalProcessorContext<KOut, VOut> context, final ProcessingExceptionHandler processingExceptionHandler) {
+        init(context);
+        this.processingExceptionHandler = processingExceptionHandler;
+    }
+
     public void close() {
         throwIfClosed();
 
@@ -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();
+            }
         }
     }
 

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.processor.internals;
 
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeader;
@@ -37,17 +38,28 @@ public class ProcessorRecordContext implements RecordContext, RecordMetadata {
     private final String topic;
     private final int partition;
     private final Headers headers;
+    private final ConsumerRecord<byte[], byte[]> rawRecord;
 
     public ProcessorRecordContext(final long timestamp,
                                   final long offset,
                                   final int partition,
                                   final String topic,
                                   final Headers headers) {
+        this(timestamp, offset, partition, topic, headers, null);
+    }
+
+    public ProcessorRecordContext(final long timestamp,
+                                  final long offset,
+                                  final int partition,
+                                  final String topic,
+                                  final Headers headers,
+                                  final ConsumerRecord<byte[], byte[]> rawRecord) {
         this.timestamp = timestamp;
         this.offset = offset;
         this.topic = topic;
         this.partition = partition;
         this.headers = Objects.requireNonNull(headers);
+        this.rawRecord = rawRecord;
     }
 
     @Override
@@ -75,6 +87,10 @@ public Headers headers() {
         return headers;
     }
 
+    public ConsumerRecord<byte[], byte[]> rawRecord() {
+        return rawRecord;
+    }
+
     public long residentMemorySizeEstimate() {
         long size = 0;
         size += Long.BYTES; // value.context.timestamp
@@ -173,7 +189,7 @@ public static ProcessorRecordContext deserialize(final ByteBuffer buffer) {
             headers = new RecordHeaders(headerArr);
         }
 
-        return new ProcessorRecordContext(timestamp, offset, partition, topic, headers);
+        return new ProcessorRecordContext(timestamp, offset, partition, topic, headers, null);
     }
 
     @Override

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

@@ -230,7 +230,7 @@ private void updateHead() {
                 droppedRecordsSensor.record();
                 continue;
             }
-            headRecord = new StampedRecord(deserialized, timestamp);
+            headRecord = new StampedRecord(deserialized, timestamp, raw);
             headRecordSizeInBytes = consumerRecordSizeInBytes(raw);
         }
 

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

@@ -88,7 +88,8 @@ public void process(final Record<KIn, VIn> record) {
                 context.offset(),
                 context.partition(),
                 context.topic(),
-                record.headers()
+                record.headers(),
+                context.recordContext().rawRecord()
             );
 
         final String topic = topicExtractor.extract(key, value, contextForExtraction);

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

@@ -20,9 +20,11 @@
 import org.apache.kafka.common.header.Headers;
 
 public class StampedRecord extends Stamped<ConsumerRecord<?, ?>> {
+    private final ConsumerRecord<byte[], byte[]> rawRecord;
 
-    public StampedRecord(final ConsumerRecord<?, ?> record, final long timestamp) {
+    public StampedRecord(final ConsumerRecord<?, ?> record, final long timestamp, final ConsumerRecord<byte[], byte[]> rawRecord) {
         super(record, timestamp);
+        this.rawRecord = rawRecord;
     }
 
     public String topic() {
@@ -49,6 +51,20 @@ public Headers headers() {
         return value.headers();
     }
 
+    public ConsumerRecord<byte[], byte[]> rawRecord() {
+        return rawRecord;
+    }
+
+    @Override
+    public boolean equals(final Object other) {
+        return super.equals(other);
+    }
+
+    @Override
+    public int hashCode() {
+        return super.hashCode();
+    }
+
     @Override
     public String toString() {
         return value.toString() + ", timestamp = " + timestamp;