任何人都知道如何在不手动创建ConcurrentKafkaListenerContainerFactory的情况下设置新属性: authorizationExceptionRetryInterval。
发布于 2020-02-04 22:04:52
我想说的是..。
@Component
class ContainerFactoryCustomizer {
ContainerFactoryCustomizer(AbstractKafkaListenerContainerFactory<?, ?, ?> factory) {
factory.setContainerCustomizer(
container -> container.getContainerProperties()
.setAuthorizationExceptionRetryInterval(Duration.ofSeconds(10L)));
}
}
但那不管用,due to a bug (the container customizer is not set up)。
以下是一种解决方法:
@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;修复属性并启动容器)。
https://stackoverflow.com/questions/60054097
复制相似问题