From 596f7d940caa2f562775d3f232be4ce7b58bffe9 Mon Sep 17 00:00:00 2001
From: Filip Hrisafov <filip.hrisafov@gmail.com>
Date: Tue, 21 Jun 2022 22:32:40 +0200
Subject: [PATCH] Widen Delegate Type for KafkaBackoffAwareMLAdapter

Setting up the backoff aware message listener adapter should work for any message listener.
---
 .../retrytopic/ListenerContainerFactoryConfigurer.java      | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java
index ffa5e012..9230a9e4 100644
--- a/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java
+++ b/spring-kafka/src/main/java/org/springframework/kafka/retrytopic/ListenerContainerFactoryConfigurer.java
@@ -33,13 +33,13 @@ import org.springframework.kafka.KafkaException;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerEndpoint;
-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.DefaultErrorHandler;
 import org.springframework.kafka.listener.KafkaConsumerBackoffManager;
+import org.springframework.kafka.listener.MessageListener;
 import org.springframework.kafka.listener.adapter.KafkaBackoffAwareMessageListenerAdapter;
 import org.springframework.kafka.support.TopicPartitionOffset;
 import org.springframework.util.Assert;
@@ -252,8 +252,8 @@ public class ListenerContainerFactoryConfigurer {
 
 	protected void setupBackoffAwareMessageListenerAdapter(ConcurrentMessageListenerContainer<?, ?> container,
 														Configuration configuration, boolean isSetContainerProperties) {
-		AcknowledgingConsumerAwareMessageListener<?, ?> listener = checkAndCast(container.getContainerProperties()
-				.getMessageListener(), AcknowledgingConsumerAwareMessageListener.class);
+		MessageListener<?, ?> listener = checkAndCast(container.getContainerProperties()
+				.getMessageListener(), MessageListener.class);
 
 		if (isSetContainerProperties && !configuration.backOffValues.isEmpty()) {
 			configurePollTimeoutAndIdlePartitionInterval(container, configuration);