Deprecate Accessors for Legacy Error Handlers

**cherry-pick to 2.8.x**

Author: garyrussell

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2021 the original author or authors.
+ * Copyright 2014-2022 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.
@@ -253,7 +253,10 @@ public void setReplyTemplate(KafkaTemplate<?, ?> replyTemplate) {
 	 * Set the error handler to call when the listener throws an exception.
 	 * @param errorHandler the error handler.
 	 * @since 2.2
+	 * @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
+	 * @see #setCommonErrorHandler(CommonErrorHandler)
 	 */
+	@Deprecated
 	public void setErrorHandler(ErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
@@ -262,7 +265,10 @@ public void setErrorHandler(ErrorHandler errorHandler) {
 	 * Set the batch error handler to call when the listener throws an exception.
 	 * @param errorHandler the error handler.
 	 * @since 2.2
+	 * @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
+	 * @see #setCommonErrorHandler(CommonErrorHandler)
 	 */
+	@Deprecated
 	public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
@@ -419,6 +425,7 @@ private void configureEndpoint(AbstractKafkaListenerEndpoint<K, V> aklEndpoint)
 	 * @param instance the container instance to configure.
 	 * @param endpoint the endpoint.
 	 */
+	@SuppressWarnings("deprecation")
 	protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
 		ContainerProperties properties = instance.getContainerProperties();
 		BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",

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

@@ -207,7 +207,10 @@ public ApplicationEventPublisher getApplicationEventPublisher() {
 	 * Set the error handler to call when the listener throws an exception.
 	 * @param errorHandler the error handler.
 	 * @since 2.2
+	 * @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
+	 * @see #setCommonErrorHandler(CommonErrorHandler)
 	 */
+	@Deprecated
 	public void setErrorHandler(ErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
@@ -216,16 +219,22 @@ public void setErrorHandler(ErrorHandler errorHandler) {
 	 * Set the error handler to call when the listener throws an exception.
 	 * @param errorHandler the error handler.
 	 * @since 2.2
+	 * @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
+	 * @see #setCommonErrorHandler(CommonErrorHandler)
 	 */
-	public void setGenericErrorHandler(GenericErrorHandler<?> errorHandler) {
+	@Deprecated
+	public void setGenericErrorHandler(@Nullable GenericErrorHandler<?> errorHandler) {
 		this.errorHandler = errorHandler;
 	}
 
 	/**
 	 * Set the batch error handler to call when the listener throws an exception.
 	 * @param errorHandler the error handler.
 	 * @since 2.2
+	 * @deprecated in favor of {@link #setCommonErrorHandler(CommonErrorHandler)}
+	 * @see #setCommonErrorHandler(CommonErrorHandler)
 	 */
+	@Deprecated
 	public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
 		this.errorHandler = errorHandler;
 	}
@@ -234,7 +243,10 @@ public void setBatchErrorHandler(BatchErrorHandler errorHandler) {
 	 * Get the configured error handler.
 	 * @return the error handler.
 	 * @since 2.2
+	 * @deprecated in favor of {@link #getCommonErrorHandler()}
+	 * @see #getCommonErrorHandler()
 	 */
+	@Deprecated
 	@Nullable
 	public GenericErrorHandler<?> getGenericErrorHandler() {
 		return this.errorHandler;

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-2022 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.
@@ -212,6 +212,7 @@ protected void doStart() {
 		}
 	}
 
+	@SuppressWarnings("deprecation")
 	private void configureChildContainer(int index, KafkaMessageListenerContainer<K, V> container) {
 		String beanName = getBeanName();
 		beanName = (beanName == null ? "consumer" : beanName) + "-" + index;

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

@@ -768,7 +768,6 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 			this.commitCurrentOnAssignment = determineCommitCurrent(consumerProperties,
 					KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties());
 			subscribeOrAssignTopics(this.consumer);
-			GenericErrorHandler<?> errHandler = getGenericErrorHandler();
 			if (listener instanceof BatchMessageListener) {
 				this.listener = null;
 				this.batchListener = (BatchMessageListener<K, V>) listener;
@@ -792,7 +791,7 @@ else if (listener instanceof MessageListener) {
 			this.listenerType = listenerType;
 			this.isConsumerAwareListener = listenerType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE)
 					|| listenerType.equals(ListenerType.CONSUMER_AWARE);
-			this.commonErrorHandler = determineCommonErrorHandler(errHandler);
+			this.commonErrorHandler = determineCommonErrorHandler();
 			Assert.state(!this.isBatchListener || !this.isRecordAck,
 					"Cannot use AckMode.RECORD with a batch listener");
 			if (this.containerProperties.getScheduler() != null) {
@@ -848,8 +847,10 @@ else if (this.commonRecordInterceptor != null) {
 		}
 
 		@Nullable
-		private CommonErrorHandler determineCommonErrorHandler(@Nullable GenericErrorHandler<?> errHandler) {
+		private CommonErrorHandler determineCommonErrorHandler() {
 			CommonErrorHandler common = getCommonErrorHandler();
+			@SuppressWarnings("deprecation")
+			GenericErrorHandler<?> errHandler = getGenericErrorHandler();
 			if (common != null) {
 				if (errHandler != null) {
 					this.logger.debug("GenericErrorHandler is ignored when a CommonErrorHandler is provided");
@@ -860,7 +861,7 @@ private CommonErrorHandler determineCommonErrorHandler(@Nullable GenericErrorHan
 				return new DefaultErrorHandler();
 			}
 			if (this.isBatchListener) {
-				validateErrorHandler(true);
+				validateErrorHandler(true, errHandler);
 				BatchErrorHandler batchErrorHandler = (BatchErrorHandler) errHandler;
 				if (batchErrorHandler != null) {
 					return new ErrorHandlerAdapter(batchErrorHandler);
@@ -870,7 +871,7 @@ private CommonErrorHandler determineCommonErrorHandler(@Nullable GenericErrorHan
 				}
 			}
 			else {
-				validateErrorHandler(false);
+				validateErrorHandler(false, errHandler);
 				ErrorHandler eh = (ErrorHandler) errHandler;
 				if (eh != null) {
 					return new ErrorHandlerAdapter(eh);
@@ -1201,8 +1202,7 @@ private void seekPartitions(Collection<TopicPartition> partitions, boolean idle)
 			}
 		}
 
-		private void validateErrorHandler(boolean batch) {
-			GenericErrorHandler<?> errHandler = KafkaMessageListenerContainer.this.getGenericErrorHandler();
+		private void validateErrorHandler(boolean batch, @Nullable GenericErrorHandler<?> errHandler) {
 			if (errHandler == null) {
 				return;
 			}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -1034,6 +1034,7 @@ public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>
 		}
 
 		@Bean
+		@SuppressWarnings("deprecation")
 		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
 				factoryWithBadConverter() {
 
@@ -1269,6 +1270,7 @@ public KafkaListenerContainerFactory<?> batchManualFactory2() {
 		}
 
 		@Bean
+		@SuppressWarnings("deprecation")
 		public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>>
 				recordAckListenerContainerFactory() {
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2021 the original author or authors.
+ * Copyright 2019-2022 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.
@@ -124,7 +124,7 @@ else if (event instanceof ConsumerFailedToStartEvent) {
 		exec.destroy();
 	}
 
-	@SuppressWarnings({ "rawtypes", "unchecked" })
+	@SuppressWarnings({ "rawtypes", "unchecked", "deprecation" })
 	@Test
 	void testCorrectContainerForConsumerError() throws InterruptedException {
 		ConsumerFactory consumerFactory = mock(ConsumerFactory.class);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -595,6 +595,7 @@ public ConsumerRecords<Integer, String> answer(InvocationOnMock invocation) thro
 		container.stop();
 	}
 
+	@SuppressWarnings("deprecation")
 	@Test
 	public void testListenerException() throws Exception {
 		this.logger.info("Start exception");

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2022 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.
@@ -151,6 +151,7 @@ public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerCont
 			return factory(cfWithExplicitDeserializers());
 		}
 
+		@SuppressWarnings("deprecation")
 		private ConcurrentKafkaListenerContainerFactory<String, String> factory(ConsumerFactory<String, String> cf) {
 			ConcurrentKafkaListenerContainerFactory<String, String> factory =
 					new ConcurrentKafkaListenerContainerFactory<>();

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

@@ -1363,6 +1363,7 @@ else if (entry.getValue().offset() == 2) {
 		logger.info("Stop batch listener manual");
 	}
 
+	@SuppressWarnings("deprecation")
 	@Test
 	public void testBatchListenerErrors() throws Exception {
 		logger.info("Start batch listener errors");
@@ -1430,7 +1431,7 @@ public void testBatchListenerErrors() throws Exception {
 		logger.info("Stop batch listener errors");
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "deprecation" })
 	@Test
 	public void testBatchListenerAckAfterRecoveryMock() throws Exception {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
@@ -2413,6 +2414,7 @@ public void testBadAckMode() {
 	}
 
 	@Test
+	@SuppressWarnings("deprecation")
 	public void testBadErrorHandler() {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka);
 		DefaultKafkaConsumerFactory<Integer, Foo1> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -2430,6 +2432,7 @@ public void testBadErrorHandler() {
 	}
 
 	@Test
+	@SuppressWarnings("deprecation")
 	public void testBadBatchErrorHandler() {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka);
 		DefaultKafkaConsumerFactory<Integer, Foo1> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -2977,7 +2980,7 @@ public void testAckModeCount() throws Exception {
 		container.stop();
 	}
 
-	@SuppressWarnings({ "unchecked", "rawtypes" })
+	@SuppressWarnings({ "unchecked", "rawtypes", "deprecation" })
 	@Test
 	public void testCommitErrorHandlerCalled() throws Exception {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2019-2021 the original author or authors.
+ * Copyright 2019-2022 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.
@@ -194,7 +194,7 @@ public Consumer consumer() {
 			return consumer;
 		}
 
-		@SuppressWarnings({ "rawtypes", "unchecked" })
+		@SuppressWarnings({ "rawtypes", "unchecked", "deprecation" })
 		@Bean
 		public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2022 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.
@@ -77,6 +77,7 @@ public static void setup() {
 		embeddedKafka = EmbeddedKafkaCondition.getBroker();
 	}
 
+	@SuppressWarnings("deprecation")
 	@Test
 	public void testRetriesAndDlt() throws InterruptedException {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("retryBatch", "false", embeddedKafka);
@@ -143,6 +144,7 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
 		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
+	@SuppressWarnings("deprecation")
 	@Test
 	public void testRetriesCantRecover() throws InterruptedException {
 		Map<String, Object> props = KafkaTestUtils.consumerProps("retryBatch2", "false", embeddedKafka);
@@ -213,7 +215,7 @@ public void accept(ConsumerRecord<?, ?> record, Exception exception) {
 		assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue();
 	}
 
-	@SuppressWarnings("unchecked")
+	@SuppressWarnings({ "unchecked", "deprecation" })
 	@Test
 	void consumerEx() throws InterruptedException {
 		ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);

spring-kafka/src/test/kotlin/org/springframework/kafka/listener/EnableKafkaKotlinTests.kt

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2021 the original author or authors.
+ * Copyright 2016-2022 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.
@@ -151,6 +151,7 @@ class EnableKafkaKotlinTests {
 		}
 
 		@Bean
+		@Suppress("deprecation")
 		fun kafkaListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, String> {
 			val factory: ConcurrentKafkaListenerContainerFactory<String, String>
 				= ConcurrentKafkaListenerContainerFactory()
@@ -167,6 +168,7 @@ class EnableKafkaKotlinTests {
 		}
 
 		@Bean
+		@Suppress("deprecation")
 		fun kafkaBatchListenerContainerFactory(): ConcurrentKafkaListenerContainerFactory<String, String> {
 			val factory: ConcurrentKafkaListenerContainerFactory<String, String>
 					= ConcurrentKafkaListenerContainerFactory()