spring-projectsGH-615: Complete CommonErrorHandler Work

Resolves spring-projects#615

- Deprecate remaining legacy error handlers
- Fix `DefaultErrorHandler` for remaining records and delivery header
- Docs

Author: garyrussell

spring-kafka-docs/src/main/asciidoc/kafka.adoc

@@ -20,3 +20,9 @@ See <<ooo-commits>> for more information.
 
 You can now receive a single record, given the topic, partition and offset.
 See <<kafka-template-receive>> for more information.
+
+[[x28-eh]]
+==== `CommonErrorHandler` Added
+
+The legacy `GenericErrorHandler` and its sub-interface hierarchies for record an batch listeners have been replaced by a new single interface `CommonErrorHandler` with implementations corresponding to most legacy implementations of `GenericErrorHandler`.
+See <<error-handlers>> for more information.

spring-kafka-docs/src/main/asciidoc/whats-new.adoc

@@ -116,6 +116,16 @@ public void setAckAfterHandle(boolean ackAfterHandle) {
 		this.ackAfterHandle = ackAfterHandle;
 	}
 
+	@Override
+	public boolean remainingRecords() {
+		return true;
+	}
+
+	@Override
+	public boolean deliveryAttemptHeader() {
+		return true;
+	}
+
 	@Override
 	public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
 			Consumer<?, ?> consumer, MessageListenerContainer container) {

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

@@ -810,7 +810,7 @@ private CommonErrorHandler determineCommonErrorHandler(@Nullable GenericErrorHan
 			}
 			if (this.isBatchListener) {
 				validateErrorHandler(true);
-				BatchErrorHandler batchErrorHandler = determineBatchErrorHandler(errHandler);
+				BatchErrorHandler batchErrorHandler = (BatchErrorHandler) errHandler;
 				if (batchErrorHandler != null) {
 					return new ErrorHandlerAdapter(batchErrorHandler);
 				}
@@ -820,7 +820,7 @@ private CommonErrorHandler determineCommonErrorHandler(@Nullable GenericErrorHan
 			}
 			else {
 				validateErrorHandler(false);
-				ErrorHandler eh = determineErrorHandler(errHandler);
+				ErrorHandler eh = (ErrorHandler) errHandler;
 				if (eh != null) {
 					return new ErrorHandlerAdapter(eh);
 				}
@@ -1120,18 +1120,6 @@ protected void checkConsumer() {
 			}
 		}
 
-		@Nullable
-		protected BatchErrorHandler determineBatchErrorHandler(@Nullable GenericErrorHandler<?> errHandler) {
-			return errHandler != null ? (BatchErrorHandler) errHandler
-					: this.transactionManager != null ? null : new RecoveringBatchErrorHandler();
-		}
-
-		@Nullable
-		protected ErrorHandler determineErrorHandler(@Nullable GenericErrorHandler<?> errHandler) {
-			return errHandler != null ? (ErrorHandler) errHandler
-					: this.transactionManager != null ? null : new SeekToCurrentErrorHandler();
-		}
-
 		@Nullable
 		private MicrometerHolder obtainMicrometerHolder() {
 			MicrometerHolder holder = null;

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

@@ -35,11 +35,14 @@
  * exception, error handling is delegated to a {@link SeekToCurrentBatchErrorHandler} with
  * this handler's {@link BackOff}. If the record is recovered, its offset is committed.
  *
+ * @deprecated in favor of {@link DefaultErrorHandler}.
+ *
  * @author Gary Russell
  * @author Myeonghyeon Lee
  * @since 2.5
  *
  */
+@Deprecated
 public class RecoveringBatchErrorHandler extends FailedBatchProcessor
 		implements ContainerAwareBatchErrorHandler {
 

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

@@ -48,6 +48,7 @@ public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware
 
 	private final BiConsumer<ConsumerRecords<?, ?>, Exception> recoverer;
 
+	@SuppressWarnings("deprecation")
 	private final CommonErrorHandler seeker = new ErrorHandlerAdapter(new SeekToCurrentBatchErrorHandler());
 
 	private boolean ackAfterHandle = true;

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

@@ -31,10 +31,14 @@
  * An error handler that seeks to the current offset for each topic in a batch of records.
  * Used to rewind partitions after a message failure so that the batch can be replayed.
  *
+ * @deprecated with no replacement - use {@link DefaultErrorHandler} with an infinite
+ * {@link BackOff}.
+ *
  * @author Gary Russell
  * @since 2.1
  *
  */
+@Deprecated
 public class SeekToCurrentBatchErrorHandler extends KafkaExceptionLogLevelAware
 		implements ContainerAwareBatchErrorHandler {
 

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

@@ -30,12 +30,15 @@
  * records. Used to rewind partitions after a message failure so that it can be
  * replayed.
  *
+ * @deprecated in favor of {@link DefaultErrorHandler}.
+ *
  * @author Gary Russell
  * @author Artem Bilan
  *
  * @since 2.0.1
  *
  */
+@Deprecated
 public class SeekToCurrentErrorHandler extends FailedRecordProcessor implements ContainerAwareErrorHandler {
 
 	private boolean ackAfterHandle = true;

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

@@ -30,20 +30,20 @@
 import org.springframework.core.log.LogAccessor;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.listener.AcknowledgingConsumerAwareMessageListener;
+import org.springframework.kafka.listener.CommonErrorHandler;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.ContainerProperties;
 import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
-import org.springframework.kafka.listener.ErrorHandler;
+import org.springframework.kafka.listener.DefaultErrorHandler;
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
-import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
 import org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter;
 import org.springframework.util.Assert;
 import org.springframework.util.backoff.FixedBackOff;
 
 /**
  *
  * Configures the provided {@link ConcurrentKafkaListenerContainerFactory} with a
- * {@link SeekToCurrentErrorHandler}, the {@link DeadLetterPublishingRecoverer} created by
+ * {@link DefaultErrorHandler}, the {@link DeadLetterPublishingRecoverer} created by
  * the {@link DeadLetterPublishingRecovererFactory}.
  *
  * Mind that the same factory can be used by many different
@@ -76,7 +76,7 @@ public class ListenerContainerFactoryConfigurer {
 	private Consumer<ConcurrentMessageListenerContainer<?, ?>> containerCustomizer = container -> {
 	};
 
-	private Consumer<ErrorHandler> errorHandlerCustomizer = errorHandler -> {
+	private Consumer<CommonErrorHandler> errorHandlerCustomizer = errorHandler -> {
 	};
 
 	private final DeadLetterPublishingRecovererFactory deadLetterPublishingRecovererFactory;
@@ -108,12 +108,13 @@ public class ListenerContainerFactoryConfigurer {
 				: doConfigure(containerFactory, Collections.emptyList());
 	}
 
-	private ConcurrentKafkaListenerContainerFactory<?, ?> doConfigure(ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory,
-																	List<Long> backOffValues) {
-		containerFactory.setContainerCustomizer(container ->
-				setupBackoffAwareMessageListenerAdapter(container, backOffValues));
+	private ConcurrentKafkaListenerContainerFactory<?, ?> doConfigure(
+			ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory, List<Long> backOffValues) {
+
+		containerFactory
+				.setContainerCustomizer(container -> setupBackoffAwareMessageListenerAdapter(container, backOffValues));
 		containerFactory
-				.setErrorHandler(createErrorHandler(this.deadLetterPublishingRecovererFactory.create()));
+				.setCommonErrorHandler(createErrorHandler(this.deadLetterPublishingRecovererFactory.create()));
 		return containerFactory;
 	}
 
@@ -135,12 +136,12 @@ public void setContainerCustomizer(Consumer<ConcurrentMessageListenerContainer<?
 		this.containerCustomizer = containerCustomizer;
 	}
 
-	public void setErrorHandlerCustomizer(Consumer<ErrorHandler> errorHandlerCustomizer) {
+	public void setErrorHandlerCustomizer(Consumer<CommonErrorHandler> errorHandlerCustomizer) {
 		this.errorHandlerCustomizer = errorHandlerCustomizer;
 	}
 
-	private ErrorHandler createErrorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
-		SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(deadLetterPublishingRecoverer,
+	private CommonErrorHandler createErrorHandler(DeadLetterPublishingRecoverer deadLetterPublishingRecoverer) {
+		DefaultErrorHandler errorHandler = new DefaultErrorHandler(deadLetterPublishingRecoverer,
 				new FixedBackOff(0, 0));
 		errorHandler.setCommitRecovered(true);
 		this.errorHandlerCustomizer.accept(errorHandler);

spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-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.
@@ -37,8 +37,8 @@
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
+import org.springframework.kafka.listener.DefaultErrorHandler;
 import org.springframework.kafka.listener.MessageListenerContainer;
-import org.springframework.kafka.listener.SeekToCurrentErrorHandler;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
@@ -88,18 +88,18 @@ public KafkaListenerContainerFactory<?> kafkaListenerContainerFactory(EmbeddedKa
 			ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();
 			factory.setConsumerFactory(consumerFactory(embeddedKafka));
-			SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler() {
+			DefaultErrorHandler errorHandler = new DefaultErrorHandler() {
 
 				@Override
-				public void handle(Exception thrownException, List<ConsumerRecord<?, ?>> records,
+				public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records,
 						Consumer<?, ?> consumer, MessageListenerContainer container) {
 					Config.this.seekPerformed = true;
-					super.handle(thrownException, records, consumer, container);
+					super.handleRemaining(thrownException, records, consumer, container);
 				}
 
 			};
 			errorHandler.setLogLevel(Level.INFO);
-			factory.setErrorHandler(errorHandler);
+			factory.setCommonErrorHandler(errorHandler);
 			factory.setStatefulRetry(true);
 			factory.setRetryTemplate(new RetryTemplate());
 			factory.setRecoveryCallback(c -> {

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

@@ -917,8 +917,8 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		containerProps.setClientId("clientId");
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
-		SeekToCurrentErrorHandler errorHandler = spy(new SeekToCurrentErrorHandler(new FixedBackOff(0L, 0)));
-		container.setErrorHandler(errorHandler);
+		DefaultErrorHandler errorHandler = spy(new DefaultErrorHandler(new FixedBackOff(0L, 0)));
+		container.setCommonErrorHandler(errorHandler);
 		container.start();
 		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -927,7 +927,7 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
 		inOrder.verify(consumer).commitSync(anyMap(), any());
 		inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
-		inOrder.verify(errorHandler).handle(any(), any(), any(), any());
+		inOrder.verify(errorHandler).handleRemaining(any(), any(), any(), any());
 		inOrder.verify(consumer).commitSync(anyMap(), any());
 		container.stop();
 	}
@@ -2879,7 +2879,6 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.setBeanName("testContainerException");
-		container.setErrorHandler(new SeekToCurrentErrorHandler());
 		container.start();
 		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
 		template.sendDefault(0, 0, "a");
@@ -3438,7 +3437,7 @@ void commitAfterHandleManual() throws InterruptedException {
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		AtomicBoolean recovered = new AtomicBoolean();
 		CountDownLatch latch = new CountDownLatch(1);
-		container.setErrorHandler(new SeekToCurrentErrorHandler((rec, ex) -> {
+		container.setCommonErrorHandler(new DefaultErrorHandler((rec, ex) -> {
 					recovered.set(true);
 					latch.countDown();
 				},

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-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.
@@ -200,7 +200,6 @@ public Consumer consumer() {
 		public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
 			factory.setConsumerFactory(consumerFactory());
-			factory.setErrorHandler(new SeekToCurrentErrorHandler());
 			factory.getContainerProperties().setAckMode(AckMode.RECORD);
 			factory.getContainerProperties().setDeliveryAttemptHeader(true);
 			return factory;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-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.
@@ -209,7 +209,6 @@ public Consumer consumer() {
 		public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
 			factory.setConsumerFactory(consumerFactory());
-			factory.setErrorHandler(new SeekToCurrentErrorHandler());
 			factory.getContainerProperties().setAckMode(AckMode.MANUAL);
 			factory.getContainerProperties().setMissingTopicsFatal(false);
 			return factory;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-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.
@@ -74,6 +74,7 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
+@SuppressWarnings("deprecation")
 public class ManualNackRecordTxTests {
 
 	@SuppressWarnings("rawtypes")

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

@@ -48,6 +48,7 @@
  * @since 2.5
  *
  */
+@SuppressWarnings("deprecation")
 @EmbeddedKafka(topics = {
 		RecoveringBatchErrorHandlerIntegrationTests.topic1,
 		RecoveringBatchErrorHandlerIntegrationTests.topic1DLT,

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

@@ -70,6 +70,7 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
+@SuppressWarnings("deprecation")
 public class RecoveringBatchErrorHandlerTests {
 
 	private static final String CONTAINER_ID = "container";

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-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.
@@ -79,6 +79,7 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
+@SuppressWarnings("deprecation")
 public class SeekToCurrentBatchErrorHandlerTests {
 
 	private static final String CONTAINER_ID = "container";

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

@@ -48,6 +48,7 @@
  * @since 2.3
  *
  */
+@SuppressWarnings("deprecation")
 public class SeekToCurrentErrorHandlerTests {
 
 	@Test

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-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.
@@ -72,6 +72,7 @@
  */
 @SpringJUnitConfig
 @DirtiesContext
+@SuppressWarnings("deprecation")
 public class SeekToCurrentOnErrorBatchModeTXTests {
 
 	private static final String CONTAINER_ID = "container";
@@ -236,7 +237,7 @@ public Consumer consumer() {
 		public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
 			factory.setConsumerFactory(consumerFactory());
-			factory.setErrorHandler(new SeekToCurrentErrorHandler());
+			factory.setCommonErrorHandler(new DefaultErrorHandler());
 			factory.getContainerProperties().setAckMode(AckMode.BATCH);
 			factory.getContainerProperties().setTransactionManager(tm());
 			return factory;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2020 the original author or authors.
+ * Copyright 2017-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.
@@ -206,7 +206,6 @@ public Consumer consumer() {
 		public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
 			factory.setConsumerFactory(consumerFactory());
-			factory.setErrorHandler(new SeekToCurrentErrorHandler());
 			factory.getContainerProperties().setAckMode(AckMode.BATCH);
 			return factory;
 		}