spring-projectsGH-615: Add CommonLoggingErrorHandler

See spring-projects#615

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -37,6 +37,7 @@
 import org.springframework.kafka.listener.AfterRollbackProcessor;
 import org.springframework.kafka.listener.BatchErrorHandler;
 import org.springframework.kafka.listener.BatchInterceptor;
+import org.springframework.kafka.listener.CommonErrorHandler;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.ErrorHandler;
 import org.springframework.kafka.listener.GenericErrorHandler;
@@ -75,6 +76,8 @@ public abstract class AbstractKafkaListenerContainerFactory<C extends AbstractMe
 
 	private GenericErrorHandler<?> errorHandler;
 
+	private CommonErrorHandler commonErrorHandler;
+
 	private ConsumerFactory<? super K, ? super V> consumerFactory;
 
 	private Boolean autoStartup;
@@ -262,6 +265,16 @@ public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
 
+	/**
+	 * Set the {@link CommonErrorHandler} which can handle errors for both record
+	 * and batch listeners. Replaces the use of {@link GenericErrorHandler}s.
+	 * @param commonErrorHandler the handler.
+	 * @since 2.8
+	 */
+	public void setCommonErrorHandler(CommonErrorHandler commonErrorHandler) {
+		this.commonErrorHandler = commonErrorHandler;
+	}
+
 	/**
 	 * Set a processor to invoke after a transaction rollback; typically will
 	 * seek the unprocessed topic/partition to reprocess the records.
@@ -342,7 +355,19 @@ public void setContainerCustomizer(ContainerCustomizer<K, V, C> containerCustomi
 
 	@Override
 	public void afterPropertiesSet() {
-		if (this.errorHandler != null) {
+		if (this.commonErrorHandler != null) {
+			if (Boolean.TRUE.equals(this.batchListener)) {
+				Assert.state(this.commonErrorHandler.supportsBatch(),
+						() -> "The common error handler must support batch listeners, it is a " +
+								this.commonErrorHandler.getClass().getName());
+			}
+			else {
+				Assert.state(this.commonErrorHandler.supportsRecord(),
+						() -> "The common error handler must support record listeners, it is a " +
+								this.commonErrorHandler.getClass().getName());
+			}
+		}
+		if (this.commonErrorHandler == null && this.errorHandler != null) {
 			if (Boolean.TRUE.equals(this.batchListener)) {
 				Assert.state(this.errorHandler instanceof BatchErrorHandler,
 						() -> "The error handler must be a BatchErrorHandler, not " +
@@ -412,6 +437,7 @@ protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
 				.acceptIfNotNull(this.containerProperties.getSubBatchPerPartition(),
 						properties::setSubBatchPerPartition)
 				.acceptIfNotNull(this.errorHandler, instance::setGenericErrorHandler)
+				.acceptIfNotNull(this.commonErrorHandler, instance::setCommonErrorHandler)
 				.acceptIfNotNull(this.missingTopicsFatal, instance.getContainerProperties()::setMissingTopicsFatal);
 		Boolean autoStart = endpoint.getAutoStartup();
 		if (autoStart != null) {

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

@@ -88,6 +88,8 @@ public abstract class AbstractMessageListenerContainer<K, V>
 
 	private GenericErrorHandler<?> errorHandler;
 
+	private CommonErrorHandler commonErrorHandler;
+
 	private boolean autoStartup = true;
 
 	private int phase = DEFAULT_PHASE;
@@ -236,6 +238,26 @@ public GenericErrorHandler<?> getGenericErrorHandler() {
 		return this.errorHandler;
 	}
 
+	/**
+	 * Get the {@link CommonErrorHandler}.
+	 * @return the handler.
+	 * @since 2.8
+	 */
+	@Nullable
+	public CommonErrorHandler getCommonErrorHandler() {
+		return this.commonErrorHandler;
+	}
+
+	/**
+	 * Set the {@link CommonErrorHandler} which can handle errors for both record
+	 * and batch listeners. Replaces the use of {@link GenericErrorHandler}s.
+	 * @param commonErrorHandler the handler.
+	 * @since 2.8
+	 */
+	public void setCommonErrorHandler(@Nullable CommonErrorHandler commonErrorHandler) {
+		this.commonErrorHandler = commonErrorHandler;
+	}
+
 	@Override
 	public boolean isAutoStartup() {
 		return this.autoStartup;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 the original author or authors.
+ * Copyright 2016-2021 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.
@@ -25,11 +25,14 @@
 import org.springframework.lang.Nullable;
 
 /**
- * Simple handler that invokes a {@link LoggingErrorHandler} for each record.
+ * Simple handler that logs each record.
+ *
+ * @deprecated - use the {@link CommonLoggingErrorHandler} instead.
  *
  * @author Gary Russell
  * @since 1.1
  */
+@Deprecated
 public class BatchLoggingErrorHandler implements BatchErrorHandler {
 
 	private static final LogAccessor LOGGER =
@@ -43,7 +46,7 @@ public void handle(Exception thrownException, @Nullable ConsumerRecords<?, ?> da
 		}
 		else {
 			for (ConsumerRecord<?, ?> record : data) {
-				message.append(record).append('\n');
+				message.append(ListenerUtils.recordToString(record)).append('\n');
 			}
 		}
 		LOGGER.error(thrownException, () -> message.substring(0, message.length() - 1));

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

@@ -34,12 +34,50 @@
  */
 public interface CommonErrorHandler extends DeliveryAttemptAware {
 
+	/**
+	 * The type of error handler.
+	 */
+	enum Type {
+
+		/**
+		 * Can handle errors for record listeners.
+		 */
+		RECORD,
+
+		/**
+		 * Can handle errors for batch listeners.
+		 */
+		BATCH,
+
+		/**
+		 * Can handle errors for both record and batch listeners.
+		 */
+		BOTH
+
+	}
+
 	/**
 	 * Return true if this error handler is for a batch listener.
 	 * @return true for batch.
 	 */
-	default boolean isBatch() {
-		return false;
+	default Type getType() {
+		return Type.RECORD;
+	}
+
+	/**
+	 * Return true if this handler supports record listeners.
+	 * @return true if record listeners are supported.
+	 */
+	default boolean supportsRecord() {
+		return getType().equals(Type.RECORD) || getType().equals(Type.BOTH);
+	}
+
+	/**
+	 * Return true if this handler supports batch listeners.
+	 * @return true if batch listeners are supported.
+	 */
+	default boolean supportsBatch() {
+		return getType().equals(Type.BATCH) || getType().equals(Type.BOTH);
 	}
 
 	/**

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

@@ -0,0 +1,67 @@
+/*
+ * Copyright 2015-2019 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.springframework.kafka.listener;
+
+import org.apache.commons.logging.LogFactory;
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * The {@link CommonErrorHandler} implementation for logging exceptions.
+ *
+ * @author Gary Russell
+ * @since 2.8
+ *
+ */
+public class CommonLoggingErrorHandler implements CommonErrorHandler {
+
+	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(CommonLoggingErrorHandler.class));
+
+	@Override
+	public Type getType() {
+		return Type.BOTH;
+	}
+
+	@Override
+	public void handleRecord(Exception thrownException, ConsumerRecord<?, ?> record, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+
+		LOGGER.error(thrownException, () -> "Error occured while processing: " + ListenerUtils.recordToString(record));
+	}
+
+	@Override
+	public void handleBatch(Exception thrownException, ConsumerRecords<?, ?> data, Consumer<?, ?> consumer,
+			MessageListenerContainer container, Runnable invokeListener) {
+
+		StringBuilder message = new StringBuilder("Error occurred while processing:\n");
+		for (ConsumerRecord<?, ?> record : data) {
+			message.append(ListenerUtils.recordToString(record)).append('\n');
+		}
+		LOGGER.error(thrownException, () -> message.substring(0, message.length() - 1));
+	}
+
+	@Override
+	public void handleOtherException(Exception thrownException, Consumer<?, ?> consumer,
+			MessageListenerContainer container) {
+
+		LOGGER.error(thrownException, () -> "Error occurred while not processing records");
+	}
+
+}

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

@@ -221,6 +221,7 @@ private void configureChildContainer(int index, KafkaMessageListenerContainer<K,
 		}
 		container.setClientIdSuffix(this.concurrency > 1 || this.alwaysClientIdSuffix ? "-" + index : "");
 		container.setGenericErrorHandler(getGenericErrorHandler());
+		container.setCommonErrorHandler(getCommonErrorHandler());
 		container.setAfterRollbackProcessor(getAfterRollbackProcessor());
 		container.setRecordInterceptor(getRecordInterceptor());
 		container.setBatchInterceptor(getBatchInterceptor());

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

@@ -63,8 +63,8 @@ public ErrorHandlerAdapter(BatchErrorHandler batchErrorHandler) {
 	}
 
 	@Override
-	public boolean isBatch() {
-		return this.batchErrorHandler != null;
+	public Type getType() {
+		return this.batchErrorHandler != null ? Type.BATCH : Type.RECORD;
 	}
 
 	@Override

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

@@ -734,7 +734,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 			this.commitCurrentOnAssignment = determineCommitCurrent(consumerProperties,
 					KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties());
 			subscribeOrAssignTopics(this.consumer);
-			GenericErrorHandler<?> errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler();
+			GenericErrorHandler<?> errHandler = getGenericErrorHandler();
 			if (listener instanceof BatchMessageListener) {
 				this.listener = null;
 				this.batchListener = (BatchMessageListener<K, V>) listener;
@@ -798,6 +798,15 @@ else if (listener instanceof MessageListener) {
 
 		@Nullable
 		private CommonErrorHandler determineCommonErrorHandler(GenericErrorHandler<?> errHandler) {
+			CommonErrorHandler common = getCommonErrorHandler();
+			if (common != null) {
+				if (errHandler != null) {
+					this.logger.debug("GenericErrorHandler is ignored when a CommonErrorHandler is provided");
+				}
+				Assert.state(this.isBatchListener ? common.supportsBatch() : common.supportsRecord(),
+						"Error handler is not compatible with the message listener");
+				return common;
+			}
 			if (this.isBatchListener) {
 				validateErrorHandler(true);
 				BatchErrorHandler batchErrorHandler = determineBatchErrorHandler(errHandler);
@@ -1111,13 +1120,13 @@ protected void checkConsumer() {
 		}
 
 		@Nullable
-		protected BatchErrorHandler determineBatchErrorHandler(GenericErrorHandler<?> errHandler) {
+		protected BatchErrorHandler determineBatchErrorHandler(@Nullable GenericErrorHandler<?> errHandler) {
 			return errHandler != null ? (BatchErrorHandler) errHandler
 					: this.transactionManager != null ? null : new RecoveringBatchErrorHandler();
 		}
 
 		@Nullable
-		protected ErrorHandler determineErrorHandler(GenericErrorHandler<?> errHandler) {
+		protected ErrorHandler determineErrorHandler(@Nullable GenericErrorHandler<?> errHandler) {
 			return errHandler != null ? (ErrorHandler) errHandler
 					: this.transactionManager != null ? null : new SeekToCurrentErrorHandler();
 		}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2021 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.
@@ -21,21 +21,23 @@
 
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
-import org.springframework.util.ObjectUtils;
 
 /**
  * The {@link ErrorHandler} implementation for logging purpose.
  *
+ * @deprecated - use the {@link CommonLoggingErrorHandler} instead.
+ *
  * @author Marius Bogoevici
  * @author Gary Russell
  */
+@Deprecated
 public class LoggingErrorHandler implements ErrorHandler {
 
 	private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(LoggingErrorHandler.class));
 
 	@Override
 	public void handle(Exception thrownException, @Nullable ConsumerRecord<?, ?> record) {
-		LOGGER.error(thrownException, () -> "Error while processing: " + ObjectUtils.nullSafeToString(record));
+		LOGGER.error(thrownException, () -> "Error while processing: " + ListenerUtils.recordToString(record));
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -96,6 +96,7 @@
 import org.springframework.kafka.event.ListenerContainerIdleEvent;
 import org.springframework.kafka.event.ListenerContainerNoLongerIdleEvent;
 import org.springframework.kafka.listener.AbstractConsumerSeekAware;
+import org.springframework.kafka.listener.CommonLoggingErrorHandler;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ConsumerAwareErrorHandler;
 import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
@@ -871,6 +872,9 @@ public void testKeyConversion() throws Exception {
 					.collect(Collectors.toList()), e);
 			throw e;
 		}
+		MessageListenerContainer container = this.registry.getListenerContainer("bytesKey");
+		assertThat(KafkaTestUtils.getPropertyValue(container, "commonErrorHandler"))
+				.isInstanceOf(CommonLoggingErrorHandler.class);
 	}
 
 	@Test
@@ -1107,6 +1111,7 @@ public KafkaListenerContainerFactory<?> bytesStringListenerContainerFactory() {
 				this.intercepted = true;
 				return record;
 			});
+			factory.setCommonErrorHandler(new CommonLoggingErrorHandler());
 			return factory;
 		}