GH-970: Add getGroupId() to containers

Resolves #970

- allow retrieval of the `group.id`, even if not set on the container properties
- also add `getAllListenerContainers` to the `RLERegistry` as a convenience
- also add `getListenerId` to return the id or bean name of the container

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2019 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.
@@ -112,11 +112,28 @@ public Set<String> getListenerContainerIds() {
 	/**
 	 * Return the managed {@link MessageListenerContainer} instance(s).
 	 * @return the managed {@link MessageListenerContainer} instance(s).
+	 * @see #getAllListenerContainers()
 	 */
 	public Collection<MessageListenerContainer> getListenerContainers() {
 		return Collections.unmodifiableCollection(this.listenerContainers.values());
 	}
 
+	/**
+	 * Return all {@link MessageListenerContainer} instances including those managed by
+	 * this registry and those declared as beans in the application context.
+	 * Prototype-scoped containers will be included. Lazy beans that have not yet been
+	 * created will not be initialized by a call to this method.
+	 * @return the {@link MessageListenerContainer} instance(s).
+	 * @since 2.2.5
+	 * @see #getListenerContainers()
+	 */
+	public Collection<MessageListenerContainer> getAllListenerContainers() {
+		List<MessageListenerContainer> containers = new ArrayList<>();
+		containers.addAll(getListenerContainers());
+		containers.addAll(this.applicationContext.getBeansOfType(MessageListenerContainer.class, true, false).values());
+		return containers;
+	}
+
 	/**
 	 * Create a message listener container for the given {@link KafkaListenerEndpoint}.
 	 * <p>This create the necessary infrastructure to honor that endpoint

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

@@ -37,6 +37,7 @@
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.event.ContainerStoppedEvent;
 import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
@@ -245,6 +246,21 @@ public ContainerProperties getContainerProperties() {
 		return this.containerProperties;
 	}
 
+	@Override
+	public String getGroupId() {
+		return this.containerProperties.getGroupId() == null
+				? (String) this.consumerFactory
+						.getConfigurationProperties()
+						.get(ConsumerConfig.GROUP_ID_CONFIG)
+				: this.containerProperties.getGroupId();
+	}
+
+	@Override
+	@Nullable
+	public String getListenerId() {
+		return this.beanName; // the container factory sets the bean name to the id attribute
+	}
+
 	@Override
 	public void setupMessageListener(Object messageListener) {
 		this.containerProperties.setMessageListener(messageListener);

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

@@ -448,10 +448,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
 
 		private final TransactionTemplate transactionTemplate;
 
-		private final String consumerGroupId = this.containerProperties.getGroupId() == null
-				? (String) KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties()
-				.get(ConsumerConfig.GROUP_ID_CONFIG)
-				: this.containerProperties.getGroupId();
+		private final String consumerGroupId = getGroupId();
 
 		private final TaskScheduler taskScheduler;
 

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2018 the original author or authors.
+ * Copyright 2016-2019 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.
@@ -24,6 +24,7 @@
 import org.apache.kafka.common.TopicPartition;
 
 import org.springframework.context.SmartLifecycle;
+import org.springframework.lang.Nullable;
 
 /**
  * Internal abstraction used by the framework representing a message
@@ -114,4 +115,25 @@ default void setAutoStartup(boolean autoStartup) {
 		// empty
 	}
 
+	/**
+	 * Return the {@code group.id} property for this container whether specifically set on the
+	 * container or via a consumer property on the consumer factory.
+	 * @return the group id.
+	 * @since 2.2.5
+	 */
+	default String getGroupId() {
+		throw new UnsupportedOperationException("This container does not support retrieving the group id");
+	}
+
+	/**
+	 * The 'id' attribute of a {@code @KafkaListener} or the bean name for spring-managed
+	 * containers.
+	 * @return the id or bean name.
+	 * @since 2.2.5
+	 */
+	@Nullable
+	default String getListenerId() {
+		throw new UnsupportedOperationException("This container does not support retrieving the listener id");
+	}
+
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java

@@ -27,6 +27,7 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -96,6 +97,16 @@ public void stopContainerAfterException() throws Exception {
 		inOrder.verify(this.consumer).unsubscribe();
 		inOrder.verify(this.consumer).close();
 		inOrder.verifyNoMoreInteractions();
+		assertThat(this.registry.getListenerContainers()).hasSize(1);
+		Collection<MessageListenerContainer> containers = this.registry.getAllListenerContainers();
+		assertThat(containers).hasSize(2);
+		Iterator<MessageListenerContainer> iterator = containers.iterator();
+		MessageListenerContainer one = iterator.next();
+		MessageListenerContainer two = iterator.next();
+		assertThat(one).isNotSameAs(two);
+		assertThat(two).isSameAs(this.config.springManagedContainer());
+		assertThat(one.getListenerId()).isEqualTo(CONTAINER_ID);
+		assertThat(two.getListenerId()).isEqualTo("springManagedContainer");
 	}
 
 	@Configuration
@@ -178,7 +189,7 @@ public Consumer consumer() {
 
 		@SuppressWarnings({ "rawtypes", "unchecked" })
 		@Bean
-		public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
+		public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
 			ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
 			factory.setConsumerFactory(consumerFactory());
 			factory.getContainerProperties().setAckOnError(false);
@@ -203,6 +214,13 @@ public void handle(Exception thrownException, ConsumerRecords<?, ?> records,
 			return factory;
 		}
 
+		@Bean
+		public ConcurrentMessageListenerContainer<String, String> springManagedContainer() {
+			ConcurrentMessageListenerContainer<String, String> container = kafkaListenerContainerFactory()
+					.createContainer("springManaged");
+			container.setAutoStartup(false);
+			return container;
+		}
 	}
 
 }

spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java

@@ -186,6 +186,7 @@ public void testDelegateType() throws Exception {
 						.collect(Collectors.toList()));
 			}
 		});
+		assertThat(container.getGroupId()).isEqualTo("delegate");
 		container.start();
 
 		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
@@ -270,9 +271,11 @@ public void testNoResetPolicy() throws Exception {
 			trace.set(new RuntimeException().getStackTrace());
 			latch1.countDown();
 		});
+		containerProps.setGroupId("delegateGroup");
 		KafkaMessageListenerContainer<Integer, String> container =
 				new KafkaMessageListenerContainer<>(cf, containerProps);
 		container.setBeanName("delegate");
+		assertThat(container.getGroupId()).isEqualTo("delegateGroup");
 		container.start();
 
 		int n = 0;

src/reference/asciidoc/kafka.adoc

@@ -1320,6 +1320,11 @@ private KafkaListenerEndpointRegistry registry;
 ----
 ====
 
+The registry only maintains the life cycle of containers it manages; containers declared as beans are not managed by the registry and can be obtained from the application context.
+A collection of managed containers can be obtained by calling the registry's `getListenerContainers()` method.
+Version 2.2.5 added a convenience method `getAllListenerContainers()`, which returns a collection of all containers, including those managed by the registry and those declared as beans.
+The collection returned will include any prototype beans that have been initialized, but it will not initialize any lazy bean declarations.
+
 [[kafka-validation]]
 ===== `@KafkaListener` `@Payload` Validation