GH-3319: Fix NPE in the KafkaListenerEndpointRegistry (#3320)

Fixes: #3319

The `KafkaListenerEndpointRegistry.getUnregisteredListenerContainer()`
returns `null` when container is already present in the internal `unregisteredContainers` cache

* Fix `KafkaListenerEndpointRegistry.getUnregisteredListenerContainer()` to return a container
instance from the `unregisteredContainers` cache

(cherry picked from commit 8096c9d)

Author: artembilan

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

@@ -122,7 +122,7 @@ public MessageListenerContainer getUnregisteredListenerContainer(String id) {
 			refreshContextContainers();
 			return this.unregisteredContainers.get(id);
 		}
-		return null;
+		return container;
 	}
 
 	/**

spring-kafka/src/test/java/org/springframework/kafka/config/KafkaListenerEndpointRegistryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -22,10 +22,13 @@
 
 import org.junit.jupiter.api.Test;
 
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.listener.MessageListenerContainer;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  * @since 2.8.9
  *
  */
@@ -47,4 +50,19 @@ void unregister() {
 		assertThat(unregistered).isSameAs(container);
 	}
 
+	@Test
+	void verifyUnregisteredListenerContainer() {
+		KafkaListenerEndpointRegistry registry = new KafkaListenerEndpointRegistry();
+		GenericApplicationContext applicationContext = new GenericApplicationContext();
+		ConcurrentMessageListenerContainer<?, ?> listenerContainerMock = mock(ConcurrentMessageListenerContainer.class);
+		given(listenerContainerMock.getListenerId()).willReturn("testListenerContainer");
+		applicationContext.registerBean(ConcurrentMessageListenerContainer.class, () -> listenerContainerMock);
+		applicationContext.refresh();
+		registry.setApplicationContext(applicationContext);
+		// Lazy-load from application context
+		assertThat(registry.getUnregisteredListenerContainer("testListenerContainer")).isNotNull();
+		// From internal map
+		assertThat(registry.getUnregisteredListenerContainer("testListenerContainer")).isNotNull();
+	}
+
 }