@Bean
    @ConditionalOnMissingBean(name = "kafkaTicketRegistryMessageListenerContainer")
    public ConcurrentMessageListenerContainer<String, BaseMessageQueueCommand> kafkaTicketRegistryMessageListenerContainer(
        @Qualifier("messageHandlerMethodFactory")
        final MessageHandlerMethodFactory messageHandlerMethodFactory,
        @Qualifier("messageQueueTicketRegistryReceiver")
        final QueueableTicketRegistryMessageReceiver messageQueueTicketRegistryReceiver,
        final CasConfigurationProperties casProperties,
        @Qualifier("kafkaTicketRegistryConsumerFactory")
        final ConsumerFactory<String, BaseMessageQueueCommand> kafkaTicketRegistryConsumerFactory,
        @Qualifier(TicketCatalog.BEAN_NAME)
        final TicketCatalog ticketCatalog) throws Exception {
        
        val kafka = casProperties.getTicket().getRegistry().getKafka();

        val factory = new ConcurrentKafkaListenerContainerFactory<String, BaseMessageQueueCommand>();
        factory.setConsumerFactory(kafkaTicketRegistryConsumerFactory);
        factory.setConcurrency(kafka.getConcurrency());

        val containerProperties = factory.getContainerProperties();
        containerProperties.setGroupId(kafka.getGroupId());
        
        val topics = ticketCatalog.findAll()
            .stream()
            .map(definition -> definition.getProperties().getStorageName())
            .toList()
            .toArray(String[]::new);

        val kafkaListenerEndpoint = new MethodKafkaListenerEndpoint<>();
        kafkaListenerEndpoint.setTopics(topics);
        kafkaListenerEndpoint.setGroupId(containerProperties.getGroupId());
        kafkaListenerEndpoint.setId(UUID.randomUUID().toString());

        kafkaListenerEndpoint.setBean(messageQueueTicketRegistryReceiver);
        val listenerMethod = messageQueueTicketRegistryReceiver.getClass().getMethod("receive", BaseMessageQueueCommand.class);
        kafkaListenerEndpoint.setMethod(listenerMethod);

        kafkaListenerEndpoint.setMessageHandlerMethodFactory(messageHandlerMethodFactory);
        return factory.createListenerContainer(kafkaListenerEndpoint);
    }
This is {@link CasKafkaTicketRegistryAutoConfiguration}.

@author Misagh Moayyed
@since 7.2.0
support/cas-server-support-kafka-ticket-registry/src/main/java/org/apereo/cas/config/CasKafkaTicketRegistryAutoConfiguration.java