任何人都知道如何设置新属性:authorizationExceptionRetryInterval,而无需手动创建 ConcurrentKafkaListenerContainerFactory。
我本来想说...
@Component
class ContainerFactoryCustomizer {
ContainerFactoryCustomizer(AbstractKafkaListenerContainerFactory<?, ?, ?> factory) {
factory.setContainerCustomizer(
container -> container.getContainerProperties()
.setAuthorizationExceptionRetryInterval(Duration.ofSeconds(10L)));
}
}
但这是行不通的由于错误(未设置容器定制器) https://github.com/spring-projects/spring-kafka/issues/1376.
这是一个解决方法:
@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();
};
}
}
(Set autoStartup
为假;修复属性并启动容器)。
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)