Свойство Spring Kafka для установки SeekToCurrentBatchErrorHandler
Есть ли ошибка обработки пакетов свойства spring.kafka?
spring.kafka.listener.type=BATCH и
spring.kafka.listener.ack-mode=BATCH
с SeekToCurrentBatchErrorHandler ? Заранее спасибо.
Ответы 1
Вы не можете установить его с помощью свойства, но вы можете переопределить автоматически настроенную фабрику контейнеров Boot следующим образом:
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
ConsumerFactory<Object, Object> kafkaConsumerFactory) {
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
configurer.configure(factory, kafkaConsumerFactory);
factory.setBatchErrorHandler(new SeekToCurrentBatchErrorHandler());
return factory;
}
Он получит все свойства загрузки, и вы сможете дополнительно настроить фабрику по мере необходимости.
Я добавил проблема с новой функцией для загрузки. Это не будет свойство, но оно может автоматически подключаться к @Bean, если в контексте приложения найдено ровно одно (как в настоящее время делается для непакетных ErrorHandler).
Это тоже помогает. :) Я заметил, что автосвязывание Bean-компонента ErrorHandler не совсем работает для пакета. Думал, что делаю что-то не так. :)
Не уверен, должен ли я разместить вопрос здесь. Есть ли причина, по которой функция «spring.kafka.listener.ack-on-error=false/true» не реализована?
Вы действительно не должны; здесь не любят развернутых комментариев. Просто свойств оооочень много, выбрано всего несколько, и вы всегда можете воспользоваться методикой в моем ответе для "расширенной" настройки. Вы можете открыть другую проблему, чтобы запросить ее при загрузке, но они, скорее всего, не добавят ее до версии 2.2 (в которой используется S-K 2.3, и мы изменили значение по умолчанию на false).
Спасибо Гэри - это действительно помогает. Есть ли планы добавить его в будущем?