GH-3135 : Add ListenerContainerRegistry.getListenerContainersMatchingId(Predicate)

Fixes: #3135

* Implement new `getListenerContainersMatching()` method
* Add tests
* Add @DisplayName for a test
* Add documentation
* Cover more getter documentation
* Doc : add doc for `BiPredicate` matcher
* Feature : `getListenerContainersMatching(BiPredicate<String, MessageListenerContainer> idAndContainerMatcher)`
* Doc : Remove duplication in whats'new adoc

Author: JooHyukKim

spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/kafkalistener-lifecycle.adoc

@@ -41,3 +41,49 @@ IMPORTANT: Endpoints registered after the application context has been refreshed
 An example of late registration is a bean with a `@KafkaListener` in prototype scope where an instance is created after the context is initialized.
 Starting with version 2.8.7, you can set the registry's `alwaysStartAfterRefresh` property to `false` and then the container's `autoStartup` property will define whether or not the container is started.
 
+[[retrieving-message-listener-containers]]
+
+== Retrieving MessageListenerContainers from KafkaListenerEndpointRegistry
+
+The `KafkaListenerEndpointRegistry` provides methods for retrieving `MessageListenerContainer` instances to accommodate a range of management scenarios:
+
+**All Containers**: For operations that cover all listener containers, use `getListenerContainers()` to retrieve a comprehensive collection.
+
+[source, java]
+----
+Collection<MessageListenerContainer> allContainers = registry.getListenerContainers();
+----
+
+**Specific Container by ID**: To manage an individual container, `getListenerContainer(String id)` enables retrieval by its id.
+
+[source, java]
+----
+MessageListenerContainer specificContainer = registry.getListenerContainer("myContainerId");
+----
+
+**Dynamic Container Filtering**: Introduced in version 3.2, two overloaded `getListenerContainersMatching` methods enable refined selection of containers.
+One method takes a `Predicate<String>` for ID-based filtering as a parameter, while the other takes a `BiPredicate<String, MessageListenerContainer>`
+for more advanced criteria that may include container properties or state as a parameter.
+
+[source, java]
+----
+// Prefix matching (Predicate<String>)
+Collection<MessageListenerContainer> filteredContainers =
+    registry.getListenerContainersMatching(id -> id.startsWith("productListener-retry-"));
+
+// Regex matching (Predicate<String>)
+Collection<MessageListenerContainer> regexFilteredContainers =
+    registry.getListenerContainersMatching(myPattern::matches);
+
+// Pre-built Set of IDs (Predicate<String>)
+Collection<MessageListenerContainer> setFilteredContainers =
+    registry.getListenerContainersMatching(myIdSet::contains);
+
+// Advanced Filtering: ID prefix and running state (BiPredicate<String, MessageListenerContainer>)
+Collection<MessageListenerContainer> advancedFilteredContainers =
+    registry.getListenerContainersMatching(
+        (id, container) -> id.startsWith("specificPrefix-") && container.isRunning()
+    );
+----
+
+Utilize these methods to efficiently manage and query `MessageListenerContainer` instances within your application.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -87,3 +87,12 @@ See xref:kafka/seek.adoc#seek[Seek API Docs] for more details.
 For Spring Boot applications which define an application name, this name is now used
 as a default prefix for auto-generated client IDs for certain client types.
 See xref:kafka/connecting.adoc#default-client-id-prefixes[Default client ID prefixes] for more details.
+
+[[get-listener-containers-matching]]
+== Enhanced Retrieval of MessageListenerContainers
+
+`ListenerContainerRegistry` provides two new API's dynamically find and filter `MessageListenerContainer` instances.
+`getListenerContainersMatching(Predicate<String> idMatcher)` to filter by ID and the other is
+`getListenerContainersMatching(BiPredicate<String, MessageListenerContainer> matcher)` to filter by ID and container properties.
+
+See xref:kafka/receiving-messages/kafkalistener-lifecycle.adoc#retrieving-message-listener-containers[`@KafkaListener` Lifecycle Management's API Docs] for more information.

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

@@ -25,6 +25,8 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
 
 import org.apache.commons.logging.LogFactory;
 
@@ -68,6 +70,7 @@
  * @author Gary Russell
  * @author Asi Bross
  * @author Wang Zhiyang
+ * @author Joo Hyuk Kim
  *
  * @see KafkaListenerEndpoint
  * @see MessageListenerContainer
@@ -116,6 +119,47 @@ public MessageListenerContainer getListenerContainer(String id) {
 		return this.listenerContainers.get(id);
 	}
 
+	/**
+	 * Return all {@link MessageListenerContainer} instances with id matching the predicate or
+	 * empty {@link Collection} if no such container exists.
+	 * @param idMatcher the predicate to match container id with
+	 * @return the containers or empty {@link Collection} if no container with that id exists
+	 * @since 3.2
+	 * @see #getListenerContainerIds()
+	 * @see #getListenerContainer(String)
+	 */
+	@Override
+	public Collection<MessageListenerContainer> getListenerContainersMatching(Predicate<String> idMatcher) {
+		Assert.notNull(idMatcher, "'idMatcher' cannot be null");
+		return this.listenerContainers.entrySet()
+			.stream()
+			.filter(entry -> idMatcher.test(entry.getKey()))
+			.map(Map.Entry::getValue)
+			.toList();
+	}
+
+	/**
+	 * Return all {@link MessageListenerContainer} instances that satisfy the given bi-predicate.
+	 * The {@code BiPredicate<String, MessageListenerContainer>} takes the container id and the container itself as arguments.
+	 * This allows for more sophisticated filtering, including properties or state of the container itself.
+	 * @param idAndContainerMatcher the bi-predicate to match the container id and the container
+	 * @return the containers that match the bi-predicate criteria or an empty {@link Collection} if no matching containers exist
+	 * @since 3.2
+	 * @see #getListenerContainerIds()
+	 * @see #getListenerContainersMatching(Predicate)
+	 */
+	@Override
+	public Collection<MessageListenerContainer> getListenerContainersMatching(
+		BiPredicate<String, MessageListenerContainer> idAndContainerMatcher
+	) {
+		Assert.notNull(idAndContainerMatcher, "'idAndContainerMatcher' cannot be null");
+		return this.listenerContainers.entrySet()
+			.stream()
+			.filter(entry -> idAndContainerMatcher.test(entry.getKey(), entry.getValue()))
+			.map(Map.Entry::getValue)
+			.toList();
+	}
+
 	@Override
 	@Nullable
 	public MessageListenerContainer getUnregisteredListenerContainer(String id) {

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2021-2022 the original author or authors.
+ * Copyright 2021-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.
@@ -18,13 +18,16 @@
 
 import java.util.Collection;
 import java.util.Set;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
 
 import org.springframework.lang.Nullable;
 
 /**
  * A registry for listener containers.
  *
  * @author Gary Russell
+ * @author Joo Hyuk Kim
  * @since 2.7
  *
  */
@@ -41,6 +44,30 @@ public interface ListenerContainerRegistry {
 	@Nullable
 	MessageListenerContainer getListenerContainer(String id);
 
+	/**
+	 * Return all {@link MessageListenerContainer} instances with id matching the predicate or
+	 * empty {@link Collection} if no such container exists.
+	 * @param idMatcher the predicate to match the container id with
+	 * @return the containers or empty {@link Collection} if no container with that id exists
+	 * @since 3.2
+	 * @see #getListenerContainerIds()
+	 * @see #getListenerContainer(String)
+	 */
+	Collection<MessageListenerContainer> getListenerContainersMatching(Predicate<String> idMatcher);
+
+	/**
+	 * Return all {@link MessageListenerContainer} instances that satisfy the given bi-predicate.
+	 * The {@code BiPredicate<String, MessageListenerContainer>} takes the container id and the container itself as arguments.
+	 * This allows for more sophisticated filtering, including properties or state of the container itself.
+	 * @param idAndContainerMatcher the bi-predicate to match the container id and the container
+	 * @return the containers that match the bi-predicate criteria or an empty {@link Collection} if no matching containers exist
+	 * @since 3.2
+	 * @see #getListenerContainerIds()
+	 * @see #getListenerContainersMatching(Predicate)
+	 */
+	Collection<MessageListenerContainer> getListenerContainersMatching(
+		BiPredicate<String, MessageListenerContainer> idAndContainerMatcher);
+
 	/**
 	 * Return the {@link MessageListenerContainer} with the specified id or {@code null}
 	 * if no such container exists. Returns containers that are not registered with the

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.
@@ -17,17 +17,30 @@
 package org.springframework.kafka.config;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.function.BiPredicate;
+import java.util.function.Predicate;
+import java.util.stream.Stream;
+
+import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import org.springframework.kafka.listener.MessageListenerContainer;
 
 /**
  * @author Gary Russell
+ * @author Joo Hyuk Kim
  * @since 2.8.9
- *
  */
 public class KafkaListenerEndpointRegistryTests {
 
@@ -47,4 +60,124 @@ void unregister() {
 		assertThat(unregistered).isSameAs(container);
 	}
 
+	@DisplayName("getListenerContainersMatching throws on null predicate")
+	@Test
+	void getListenerContainersMatchingThrowsOnNullPredicate() {
+		// Given
+		KafkaListenerEndpointRegistry registry = new KafkaListenerEndpointRegistry();
+		// When & Then
+		assertThatIllegalArgumentException()
+			.isThrownBy(() -> registry.getListenerContainersMatching((Predicate<String>) null))
+			.withMessage("'idMatcher' cannot be null");
+	}
+
+
+	@DisplayName("getListenerContainersMatching with BiPredicate throws on null biPredicate")
+	@Test
+	void getListenerContainersMatchingBiPredicateThrowsOnNullBiPredicate() {
+		// Given
+		KafkaListenerEndpointRegistry registry = new KafkaListenerEndpointRegistry();
+		// When & Then
+		assertThatIllegalArgumentException()
+			.isThrownBy(() -> registry.getListenerContainersMatching((BiPredicate<String, MessageListenerContainer>) null))
+			.withMessage("'idAndContainerMatcher' cannot be null");
+	}
+
+	@DisplayName("getListenerContainersMatching should return unmodifiable list")
+	@Test
+	void testGetListenerContainersMatchingReturnsUnmodifiableList() {
+		// Given
+		KafkaListenerEndpointRegistry registry = new KafkaListenerEndpointRegistry();
+		registerListenerWithId(registry, "foo");
+		// When
+		Collection<MessageListenerContainer> listeners = registry.getListenerContainersMatching(s -> true);
+		// Then
+		assertThatExceptionOfType(UnsupportedOperationException.class)
+			.isThrownBy(() -> listeners.add(mock(MessageListenerContainer.class)));
+	}
+
+	@ParameterizedTest(name = "getListenerContainersMatching({0}, {1}) = {2}")
+	@MethodSource("paramsForGetListenerContainersMatching")
+	void getListenerContainersMatching(List<String> names, Predicate<String> idMatcher, int expectedCount) {
+		// Given
+		KafkaListenerEndpointRegistry registry = new KafkaListenerEndpointRegistry();
+		registerWithListenerIds(registry, names);
+		// When
+		Collection<MessageListenerContainer> listeners = registry.getListenerContainersMatching(idMatcher);
+		// Then
+		assertThat(listeners).hasSize(expectedCount);
+	}
+
+	/**
+	 * Provides parameters for the getListenerContainersMatching test.
+	 * Each set of parameters includes a list of names, a predicate, and the expected count of matching containers.
+	 */
+	private static Stream<Arguments> paramsForGetListenerContainersMatching() {
+		List<String> names = List.of("foo", "bar", "baz");
+		return Stream.of(
+			// Case : Two names start with "b"
+			Arguments.of(names, (Predicate<String>) id -> id.startsWith("b"), 2),
+			// Case : One name starts with "f"
+			Arguments.of(names, (Predicate<String>) id -> id.startsWith("f"), 1),
+			// Case : Empty list
+			Arguments.of(new ArrayList<>(), (Predicate<String>) id -> id.startsWith("b"), 0),
+			// Case : All names match as the predicate always returns true
+			Arguments.of(names, (Predicate<String>) id -> true, 3),
+			// Case : No names match as the predicate always returns false
+			Arguments.of(names, (Predicate<String>) id -> false, 0)
+		);
+	}
+
+	@ParameterizedTest(name = "getListenerContainersMatching with BiPredicate for {0}, expecting {2} matches")
+	@MethodSource("paramsForGetListenerContainersMatchingBiPredicate")
+	void getListenerContainersMatchingBiPredicate(List<String> names, BiPredicate<String, MessageListenerContainer> idAndContainerMatcher, int expectedCount) {
+		// Given
+		KafkaListenerEndpointRegistry registry = new KafkaListenerEndpointRegistry();
+		registerWithListenerIds(registry, names);
+		// When
+		Collection<MessageListenerContainer> listeners = registry.getListenerContainersMatching(idAndContainerMatcher);
+		// Then
+		assertThat(listeners).hasSize(expectedCount);
+	}
+
+	/**
+	 * Provides parameters for the getListenerContainersMatchingBiPredicate test.
+	 * Each set of parameters includes a list of names, a bi-predicate, and the expected count of matching containers.
+	 */
+	private static Stream<Arguments> paramsForGetListenerContainersMatchingBiPredicate() {
+		List<String> names = List.of("foo", "bar", "baz");
+		return Stream.of(
+			// Case : Filters for names starting with "b" and containers that are "running"
+			Arguments.of(names,
+				(BiPredicate<String, MessageListenerContainer>) (id, container) -> id.startsWith("b") && container.isRunning(), 2),
+			// Case : Filters for names starting with "f" and containers that are "running"
+			Arguments.of(names,
+				(BiPredicate<String, MessageListenerContainer>) (id, container) -> id.startsWith("f") && container.isRunning(), 1),
+			// Case : Filters in an empty list of names
+			Arguments.of(new ArrayList<>(),
+				(BiPredicate<String, MessageListenerContainer>) (id, container) -> id.startsWith("b") && container.isRunning(), 0),
+			// Case : Filters where all containers are considered "running"
+			Arguments.of(names,
+				(BiPredicate<String, MessageListenerContainer>) (id, container) -> container.isRunning(), 3),
+			// Case : Filters where no containers are considered "running"
+			Arguments.of(names,
+				(BiPredicate<String, MessageListenerContainer>) (id, container) -> !container.isRunning(), 0)
+		);
+	}
+
+	private static void registerWithListenerIds(KafkaListenerEndpointRegistry registry, List<String> names) {
+		names.forEach(name -> registerListenerWithId(registry, name));
+	}
+
+	private static void registerListenerWithId(KafkaListenerEndpointRegistry registry, String id) {
+		KafkaListenerEndpoint endpoint = mock(KafkaListenerEndpoint.class);
+		@SuppressWarnings("unchecked")
+		KafkaListenerContainerFactory<MessageListenerContainer> factory = mock(KafkaListenerContainerFactory.class);
+		given(endpoint.getId()).willReturn(id);
+		MessageListenerContainer container = mock(MessageListenerContainer.class);
+		given(container.isRunning()).willReturn(true);
+		given(factory.createListenerContainer(endpoint)).willReturn(container);
+		registry.registerListenerContainer(endpoint, factory);
+	}
+
 }