Установка авторизацииExceptionRetryInterval для Spring Kafka

Question

Каждый знает, как установить новое свойство: authorizationExceptionRetryInterval, не создавая ConcurrentKafkaListenerContainerFactory вручную.

Answer 1

Я собирался сказать ...

@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; исправьте свойство и запустите контейнер).