GH-2055: Containers Must Implement DisposableBean

Resolves #2055

If context initialization fails, `Lifecycle.stop()` is not called.
Containers must be stopped from `DisposableBean` in this case.

**cherry-pick to 2.7.x, 2.6.x, 2.5.x**

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java

# Conflicts:
#	spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java
#	spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 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.
@@ -227,14 +227,7 @@ protected MessageListenerContainer createListenerContainer(KafkaListenerEndpoint
 	@Override
 	public void destroy() {
 		for (MessageListenerContainer listenerContainer : getListenerContainers()) {
-			if (listenerContainer instanceof DisposableBean) {
-				try {
-					((DisposableBean) listenerContainer).destroy();
-				}
-				catch (Exception ex) {
-					this.logger.warn(ex, "Failed to destroy message listener container");
-				}
-			}
+			listenerContainer.destroy();
 		}
 	}
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2019 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.
@@ -23,6 +23,7 @@
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.TopicPartition;
 
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.context.SmartLifecycle;
 import org.springframework.lang.Nullable;
 
@@ -34,7 +35,7 @@
  * @author Gary Russell
  * @author Vladimir Tsanev
  */
-public interface MessageListenerContainer extends SmartLifecycle {
+public interface MessageListenerContainer extends SmartLifecycle, DisposableBean {
 
 	/**
 	 * Setup the message listener to use. Throws an {@link IllegalArgumentException}
@@ -151,4 +152,9 @@ default String getListenerId() {
 		throw new UnsupportedOperationException("This container does not support retrieving the listener id");
 	}
 
+	@Override
+	default void destroy() {
+		stop();
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.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.
@@ -643,8 +643,9 @@ public void onMessage(ConsumerRecord<Integer, String> data) {
 		inOrder.verify(consumer).commitSync(anyMap(), any());
 		inOrder.verify(messageListener).onMessage(any(ConsumerRecord.class));
 		inOrder.verify(consumer).commitSync(anyMap(), any());
-		container.stop();
+		container.destroy();
 		assertThat(advised).containsExactly("one", "two", "one", "two");
+		assertThat(container.isRunning()).isFalse();
 	}
 
 	@SuppressWarnings("unchecked")