diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java b/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java index c3f75d35..41b15636 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointRegistry.java +++ b/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 class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec /** * Return the managed {@link MessageListenerContainer} instance(s). * @return the managed {@link MessageListenerContainer} instance(s). + * @see #getAllListenerContainers() */ public Collection 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 getAllListenerContainers() { + List 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}. *

This create the necessary infrastructure to honor that endpoint diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index ace123d0..b694b5f8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -37,6 +37,7 @@ import org.springframework.context.ApplicationEventPublisherAware; 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 abstract class AbstractMessageListenerContainer 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); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 12b089e7..7ee96de8 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -448,10 +448,7 @@ public class KafkaMessageListenerContainer // NOSONAR comment density 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; diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java index b0129730..d5d111ec 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java +++ b/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.MetricName; 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 @@ public interface MessageListenerContainer extends SmartLifecycle { // 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"); + } + } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java index f36f8421..8215babd 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java @@ -27,6 +27,7 @@ import java.time.Duration; 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 class ContainerStoppingBatchErrorHandlerTests { inOrder.verify(this.consumer).unsubscribe(); inOrder.verify(this.consumer).close(); inOrder.verifyNoMoreInteractions(); + assertThat(this.registry.getListenerContainers()).hasSize(1); + Collection containers = this.registry.getAllListenerContainers(); + assertThat(containers).hasSize(2); + Iterator 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 class ContainerStoppingBatchErrorHandlerTests { @SuppressWarnings({ "rawtypes", "unchecked" }) @Bean - public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory(); factory.setConsumerFactory(consumerFactory()); factory.getContainerProperties().setAckOnError(false); @@ -203,6 +214,13 @@ public class ContainerStoppingBatchErrorHandlerTests { return factory; } + @Bean + public ConcurrentMessageListenerContainer springManagedContainer() { + ConcurrentMessageListenerContainer container = kafkaListenerContainerFactory() + .createContainer("springManaged"); + container.setAutoStartup(false); + return container; + } } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 2d2e45b2..e69ed227 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -186,6 +186,7 @@ public class KafkaMessageListenerContainerTests { .collect(Collectors.toList())); } }); + assertThat(container.getGroupId()).isEqualTo("delegate"); container.start(); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); @@ -270,9 +271,11 @@ public class KafkaMessageListenerContainerTests { trace.set(new RuntimeException().getStackTrace()); latch1.countDown(); }); + containerProps.setGroupId("delegateGroup"); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); container.setBeanName("delegate"); + assertThat(container.getGroupId()).isEqualTo("delegateGroup"); container.start(); int n = 0; diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index ebcc3f3f..9683b945 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/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