为 Spring Kafka 设置 authorizationExceptionRetryInterval [英] Setting authorizationExceptionRetryInterval for Spring Kafka

问题描述

任何人都知道如何设置新属性:authorizationExceptionRetryInterval 而无需手动创建 ConcurrentKafkaListenerContainerFactory.

Anyone know how to set the new property: authorizationExceptionRetryInterval without creating the ConcurrentKafkaListenerContainerFactory manually.

推荐答案

我想说...

@Component
class ContainerFactoryCustomizer {

    ContainerFactoryCustomizer(AbstractKafkaListenerContainerFactory<?, ?, ?> factory) {
        factory.setContainerCustomizer(
                container -> container.getContainerProperties()
                        .setAuthorizationExceptionRetryInterval(Duration.ofSeconds(10L)));
    }

}

但这不起作用，由于错误(容器定制器未设置).

这是一个解决方法:

@SpringBootApplication
public class So60054097Application {

    public static void main(String[] args) {
        SpringApplication.run(So60054097Application.class, args);
    }

    @KafkaListener(id = "so60054097", topics = "so60054097", autoStartup = "false")
    public void listen(String in) {
        System.out.println(in);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so60054097").partitions(1).replicas(1).build();
    }


    @Bean
    public ApplicationRunner runner(KafkaListenerEndpointRegistry registry) {
        return args -> {
            MessageListenerContainer container = registry.getListenerContainer("so60054097");
            container.getContainerProperties()
                    .setAuthorizationExceptionRetryInterval(Duration.ofSeconds(10L));
            container.start();
        };
    }
}

(将 autoStartup 设置为 false；修复属性并启动容器).

(Set autoStartup to false; fix the property and start the container).

这篇关于为 Spring Kafka 设置 authorizationExceptionRetryInterval的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！