@Bean
    @RefreshScope(proxyMode = ScopedProxyMode.DEFAULT)
    public DistributedCacheManager<RegisteredService, DistributedCacheObject<RegisteredService>, PublisherIdentifier> registeredServiceDistributedCacheManager(
        final ConfigurableApplicationContext applicationContext,
        final CasConfigurationProperties casProperties,
        @Qualifier("registeredServiceDistributedCacheKafkaTopic")
        final NewTopic registeredServiceDistributedCacheKafkaTopic,
        @Qualifier("registeredServiceDistributedKafkaTemplate")
        final KafkaOperations<String, DistributedCacheObject<RegisteredService>> registeredServiceDistributedKafkaTemplate) {
        return BeanSupplier.of(DistributedCacheManager.class)
            .when(CONDITION.given(applicationContext.getEnvironment()))
            .supply(Unchecked.supplier(() -> {
                val kafka = casProperties.getServiceRegistry().getStream().getKafka();
                val factory = new KafkaObjectFactory<String, DistributedCacheObject<RegisteredService>>(kafka.getBootstrapAddress());
                try {
                    factory.getKafkaAdminClient().createTopics(List.of(registeredServiceDistributedCacheKafkaTopic)).all().get();
                } catch (final ExecutionException e) {
                    if (e.getCause() instanceof TopicExistsException) {
                        LOGGER.info(e.getMessage());
                    } else {
                        throw e;
                    }
                }
                return new RegisteredServiceKafkaDistributedCacheManager(registeredServiceDistributedKafkaTemplate, kafka.getTopic().getName());
            }))
            .otherwiseProxy()
            .get();
    }
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