@Bean
    @ConditionalOnMissingBean(name = "kafkaTicketRegistryConsumerFactory")
    public ConsumerFactory<String, BaseMessageQueueCommand> kafkaTicketRegistryConsumerFactory(
        final ConfigurableApplicationContext applicationContext,
        final CasConfigurationProperties casProperties) {
        val kafka = casProperties.getTicket().getRegistry().getKafka();
        val config = Map.<String, Object>of(
            ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapAddress(),
            ConsumerConfig.GROUP_ID_CONFIG, kafka.getGroupId()
        );
        val factory = new DefaultKafkaConsumerFactory<String, BaseMessageQueueCommand>(config);
        factory.setKeyDeserializer(new StringDeserializer());

        val valueDeserializer = new JsonDeserializer<BaseMessageQueueCommand>(MAPPER);
        valueDeserializer.addTrustedPackages(Ticket.class.getPackageName());
        valueDeserializer.addTrustedPackages(BaseMessageQueueCommand.class.getPackageName());
        valueDeserializer.addTrustedPackages(CentralAuthenticationService.NAMESPACE);
        factory.setValueDeserializer(valueDeserializer);
        
        factory.setApplicationContext(applicationContext);
        return factory;
    }
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