GH-3618: Support CommonErrorHandler for async listeners

Author: zooo-code

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

@@ -178,6 +178,7 @@
  * @author Su Ko
  * @author Jinhui Kim
  * @author Minchul Son
+ * @author Youngjoo Kim
  */
 public class KafkaMessageListenerContainer<K, V> // NOSONAR line count
 		extends AbstractMessageListenerContainer<K, V> implements ConsumerPauseResumeEventPublisher {
@@ -935,10 +936,12 @@ else if (listener instanceof MessageListener) {
 					KafkaBackoffAwareMessageListenerAdapter<K, V> genListener =
 							(KafkaBackoffAwareMessageListenerAdapter<K, V>) this.genericListener;
 					if (genListener.getDelegate() instanceof RecordMessagingMessageListenerAdapter<K, V> adapterListener) {
-						// This means that the async retry feature is supported only for SingleRecordListener with @RetryableTopic.
 						adapterListener.setCallbackForAsyncFailure(this::callbackForAsyncFailure);
 					}
 				}
+				else if (this.listener instanceof RecordMessagingMessageListenerAdapter<K, V> adapter) {
+					adapter.setCallbackForAsyncFailure(this::callbackForAsyncFailure);
+				}
 			}
 			else {
 				throw new IllegalArgumentException("Listener must be one of 'MessageListener', "
@@ -949,16 +952,6 @@ else if (listener instanceof MessageListener) {
 			this.isConsumerAwareListener = listenerType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE)
 					|| listenerType.equals(ListenerType.CONSUMER_AWARE);
 			this.commonErrorHandler = determineCommonErrorHandler();
-			// Setup async failure callback for suspend functions when CommonErrorHandler is explicitly configured
-			if (getCommonErrorHandler() != null && this.listener != null) {
-				MessageListener<?, ?> target = unwrapDelegateIfAny(this.listener);
-				if (target instanceof RecordMessagingMessageListenerAdapter<?, ?>) {
-					@SuppressWarnings("unchecked")
-					RecordMessagingMessageListenerAdapter<K, V> adapter =
-							(RecordMessagingMessageListenerAdapter<K, V>) target;
-					adapter.setCallbackForAsyncFailure(this::callbackForAsyncFailure);
-				}
-			}
 			Assert.state(!this.isBatchListener || !this.isRecordAck,
 					"Cannot use AckMode.RECORD with a batch listener");
 			if (this.containerProperties.getScheduler() != null) {

spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java

@@ -85,6 +85,7 @@
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.ContainerProperties;
+import org.springframework.kafka.listener.DefaultErrorHandler;
 import org.springframework.kafka.listener.MessageListenerContainer;
 import org.springframework.kafka.listener.RecordInterceptor;
 import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
@@ -100,6 +101,7 @@
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.util.StringUtils;
+import org.springframework.util.backoff.FixedBackOff;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
@@ -114,6 +116,7 @@
  * @author Soby Chacko
  * @author Francois Rosiere
  * @author Christian Fredriksson
+ * @author Youngjoo Kim
  *
  * @since 3.0
  */
@@ -694,6 +697,9 @@ ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerF
 					container.getContainerProperties().setAsyncAcks(true);
 					container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
 				}
+				if (container.getListenerId().equals("obs6") || container.getListenerId().equals("obs7")) {
+					container.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(0L, 0)));
+				}
 				if (container.getListenerId().equals("obs4")) {
 					container.setRecordInterceptor(new RecordInterceptor<>() {
 

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

@@ -54,6 +54,7 @@ import java.util.concurrent.TimeUnit
  *
  * @author Wang ZhiYang
  * @author Artem Bilan
+ * @author Youngjoo Kim
  *
  * @since 3.1
  */
@@ -114,7 +115,6 @@ class EnableKafkaKotlinCoroutinesTests {
 	fun `test suspend function with CommonErrorHandler`() {
 		this.template.send("kotlinAsyncTestTopicCommonHandler", "fail")
 		assertThat(this.config.commonHandlerLatch.await(10, TimeUnit.SECONDS)).isTrue()
-		assertThat(this.config.commonHandlerInvoked).isTrue()
 	}
 
 	@KafkaListener(id = "sendTopic", topics = ["kotlinAsyncTestTopic3"],
@@ -147,9 +147,6 @@ class EnableKafkaKotlinCoroutinesTests {
 		@Volatile
 		var batchError: Boolean = false
 
-		@Volatile
-		var commonHandlerInvoked: Boolean = false
-
 		val latch1 = CountDownLatch(1)
 
 		val latch2 = CountDownLatch(1)
@@ -234,7 +231,6 @@ class EnableKafkaKotlinCoroutinesTests {
 		@Bean
 		fun commonErrorHandler(): DefaultErrorHandler {
 			return DefaultErrorHandler { record, exception ->
-				commonHandlerInvoked = true
 				commonHandlerLatch.countDown()
 			}
 		}