Adding new `KafkaListenerEndpointRegistry.getListenerContainersMatchingId(predicate: Predicate<String>)`

[JooHyukKim]

Expected Behavior
Would be nice to have method...

kafkaListenerEndpointRegistry
            .getListenerContainersMatchingId( id -> id.startsWith(prefix))
           .forEach(MessageListenerContainer::start)
           

Current Behavior

kafkaListenerEndpointRegistry
            .getListenerContainerIds()
            .filter( id -> id.startsWith(prefix))
            .forEach(id -> kafkaListenerEndpointRegistry.getListenerContainer(id).start())

Context

I run application that starts/stops listeners at certain time(office hours)

We have listeners with ids such as...

• listener_product_purchase
• listener_product_purchase-retry-0
• listener_product_purchase-retry-1
• listener_product_purchase-dlt

... this, generated by @RetryableTopic configuration where I need to filter topics by some pattern and start/stop them.
Current implementation feels awkward to do iterate either nested/double.

Would be syntactic sugar.

[artembilan]

Sounds reasonable.

Any chances that you can contribute such a feature: https://github.com/spring-projects/spring-kafka/blob/main/CONTRIBUTING.adoc ?

Thanks

[JooHyukKim]

Sure 👍🏼.
Signed CLA, filed #3137, updated issue description.
LMK if you need anything else.