@Bean
    @RefreshScope(proxyMode = ScopedProxyMode.DEFAULT)
    @ConditionalOnMissingBean(name = "registeredServiceKafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<String, DistributedCacheObject> registeredServiceKafkaListenerContainerFactory(
        @Qualifier("casRegisteredServiceStreamPublisherIdentifier")
        final PublisherIdentifier casRegisteredServiceStreamPublisherIdentifier,
        final ConfigurableApplicationContext applicationContext,
        final CasConfigurationProperties casProperties) {
        val kafka = casProperties.getServiceRegistry().getStream().getKafka();
        val factory = new KafkaObjectFactory<String, DistributedCacheObject>(kafka.getBootstrapAddress());
        factory.setConsumerGroupId(casRegisteredServiceStreamPublisherIdentifier.getId());
        val mapper = new RegisteredServiceJsonSerializer(applicationContext).getObjectMapper();
        return factory.getKafkaListenerContainerFactory(new StringDeserializer(), new JsonDeserializer<>(DistributedCacheObject.class, mapper));
    }
This is {@link CasServicesStreamingKafkaAutoConfiguration}.

@author Misagh Moayyed
@since 6.3.0
support/cas-server-support-service-registry-stream-kafka/src/main/java/org/apereo/cas/config/CasServicesStreamingKafkaAutoConfiguration.java