GH-2521: Fix __listener in id Regression

Resolves #2521

The `__listener` pseudo bean name is not available for bean resolution
in the `@KafkaListener.id` attribute.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -481,10 +481,10 @@ protected void processKafkaListener(KafkaListener kafkaListener, Method method,
 		Method methodToUse = checkProxy(method, bean);
 		MethodKafkaListenerEndpoint<K, V> endpoint = new MethodKafkaListenerEndpoint<>();
 		endpoint.setMethod(methodToUse);
-		endpoint.setId(getEndpointId(kafkaListener));
 
 		String beanRef = kafkaListener.beanRef();
 		this.listenerScope.addListener(beanRef, bean);
+		endpoint.setId(getEndpointId(kafkaListener));
 		String[] topics = resolveTopics(kafkaListener);
 		TopicPartitionOffset[] tps = resolveTopicPartitions(kafkaListener);
 		if (!processMainAndRetryListeners(kafkaListener, bean, beanName, methodToUse, endpoint, topics, tps)) {

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

@@ -46,6 +46,7 @@
 import org.springframework.kafka.KafkaException.Level;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.config.KafkaListenerContainerFactory;
+import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
 import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
 import org.springframework.kafka.core.DefaultKafkaProducerFactory;
 import org.springframework.kafka.core.KafkaAdmin;
@@ -97,7 +98,8 @@ public class BatchListenerConversionTests {
 	private KafkaTemplate<Integer, Object> template;
 
 	@Test
-	public void testBatchOfPojos() throws Exception {
+	public void testBatchOfPojos(@Autowired KafkaListenerEndpointRegistry registry) throws Exception {
+		assertThat(registry.getListenerContainerIds()).contains("blc1.id", "blc2.id");
 		doTest(this.listener1, "blc1");
 		doTest(this.listener2, "blc2");
 	}
@@ -273,7 +275,8 @@ public KafkaListenerContainerFactory<?> getContainerFactory() {
 			return this.cf;
 		}
 
-		@KafkaListener(topics = "#{__listener.topic}", groupId = "#{__listener.topic}.group",
+		@KafkaListener(id = "#{__listener.topic}.id", topics = "#{__listener.topic}",
+				groupId = "#{__listener.topic}.group",
 				containerFactory = "#{__listener.containerFactory}")
 		// @SendTo("foo") test WARN log for void return
 		public void listen1(List<Foo> foos, @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics,