GH-614: Fix general consumer error handling

spring-kafka/src/main/java/org/springframework/kafka/listener/BatchErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.listener;
 
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 
 /**
@@ -29,4 +30,16 @@
  */
 public interface BatchErrorHandler extends GenericErrorHandler<ConsumerRecords<?, ?>> {
 
+	/**
+	 * Handle the exception.
+	 * @param thrownException the exception.
+	 * @param data the consumer records.
+	 * @param consumer the consumer.
+	 * @param container the container.
+	 */
+	default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+		handle(thrownException, data);
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareBatchErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -38,4 +38,10 @@ default void handle(Exception thrownException, ConsumerRecords<?, ?> data) {
 	@Override
 	void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer);
 
+	@Override
+	default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+		handle(thrownException, data, consumer);
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/ConsumerAwareErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,8 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.List;
+
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
@@ -38,4 +40,10 @@ default void handle(Exception thrownException, ConsumerRecord<?, ?> data) {
 	@Override
 	void handle(Exception thrownException, ConsumerRecord<?, ?> data, Consumer<?, ?> consumer);
 
+	@Override
+	default void handle(Exception thrownException, List<ConsumerRecord<?, ?>> data, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+		handle(thrownException, null, consumer);
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareBatchErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -35,13 +35,7 @@ default void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consu
 		throw new UnsupportedOperationException("Container should never call this");
 	}
 
-	/**
-	 * Handle the exception.
-	 * @param thrownException the exception.
-	 * @param data the consumer records.
-	 * @param consumer the consumer.
-	 * @param container the container.
-	 */
+	@Override
 	void handle(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
 			MessageListenerContainer container);
 

spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerAwareErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -39,13 +39,7 @@ default void handle(Exception thrownException, List<ConsumerRecord<?, ?>> record
 		throw new UnsupportedOperationException("Container should never call this");
 	}
 
-	/**
-	 * Handle the exception.
-	 * @param thrownException the exception.
-	 * @param records the remaining records including the one that failed.
-	 * @param consumer the consumer.
-	 * @param container the container.
-	 */
+	@Override
 	void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
 			MessageListenerContainer container);
 

spring-kafka/src/main/java/org/springframework/kafka/listener/ErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2016 the original author or authors.
+ * Copyright 2015-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,9 @@
 
 package org.springframework.kafka.listener;
 
+import java.util.List;
+
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 
 /**
@@ -26,4 +29,16 @@
  */
 public interface ErrorHandler extends GenericErrorHandler<ConsumerRecord<?, ?>> {
 
+	/**
+	 * Handle the exception.
+	 * @param thrownException the exception.
+	 * @param records the remaining records including the one that failed.
+	 * @param consumer the consumer.
+	 * @param container the container.
+	 */
+	default void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+		handle(thrownException, null);
+	}
+
 }

spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java

@@ -744,24 +744,7 @@ public void run() {
 					break;
 				}
 				catch (Exception e) {
-					try {
-						GenericErrorHandler<?> containerErrorHandler = this.containerProperties.getGenericErrorHandler();
-						if (containerErrorHandler != null) {
-							if (containerErrorHandler instanceof ConsumerAwareErrorHandler
-									|| containerErrorHandler instanceof ConsumerAwareBatchErrorHandler) {
-								containerErrorHandler.handle(e, null, this.consumer);
-							}
-							else {
-								containerErrorHandler.handle(e, null);
-							}
-						}
-						else {
-							this.logger.error("Container exception", e);
-						}
-					}
-					catch (Exception ex) {
-						this.logger.error("Container exception", ex);
-					}
+					handleConsumerException(e);
 				}
 			}
 			ProducerFactoryUtils.clearConsumerGroupId();
@@ -788,6 +771,30 @@ public void run() {
 			this.logger.info("Consumer stopped");
 		}
 
+		/**
+		 * Handle exceptions thrown by the consumer outside of message listener
+		 * invocation (e.g. commit exceptions).
+		 * @param e the exception.
+		 */
+		protected void handleConsumerException(Exception e) {
+			try {
+				if (this.errorHandler != null) {
+					this.errorHandler.handle(e, Collections.emptyList(), this.consumer,
+							KafkaMessageListenerContainer.this);
+				}
+				else if (this.batchErrorHandler != null) {
+					this.batchErrorHandler.handle(e, new ConsumerRecords<K, V>(Collections.emptyMap()), this.consumer,
+							KafkaMessageListenerContainer.this);
+				}
+				else {
+					this.logger.error("Consumer exception", e);
+				}
+			}
+			catch (Exception ex) {
+				this.logger.error("Consumer exception", ex);
+			}
+		}
+
 		private void commitPendingAcks() {
 			processCommits();
 			if (this.offsets.size() > 0) {

spring-kafka/src/main/java/org/springframework/kafka/listener/RemainingRecordsErrorHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017 the original author or authors.
+ * Copyright 2017-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -47,4 +47,10 @@ default void handle(Exception thrownException, ConsumerRecord<?, ?> data, Consum
 	 */
 	void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer);
 
+	@Override
+	default void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+		handle(thrownException, records, consumer);
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -1747,7 +1747,7 @@ public void testExceptionWhenCommitAfterRebalance() throws Exception {
 				Thread.sleep(3000);
 			}
 			catch (InterruptedException e) {
-				e.printStackTrace();
+				Thread.currentThread().interrupt();
 			}
 		});
 		containerProps.setSyncCommits(true);