@Bean
    @ConditionalOnMissingBean(name = "registeredServiceDistributedCacheKafkaTopic")
    @RefreshScope(proxyMode = ScopedProxyMode.DEFAULT)
    public NewTopic registeredServiceDistributedCacheKafkaTopic(final CasConfigurationProperties casProperties) {
        val topic = casProperties.getServiceRegistry().getStream().getKafka().getTopic();
        return TopicBuilder.name(topic.getName())
            .partitions(topic.getPartitions())
            .replicas(topic.getReplicas())
            .config(TopicConfig.COMPRESSION_TYPE_CONFIG, topic.getCompressionType())
            .compact()
            .build();
    }
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