#spring-boot #apache-kafka #spring-cloud-stream
#spring-загрузка #apache-kafka #spring-cloud-stream
Вопрос:
Я хочу установить acks = all свойство для моего продюсера в моем приложении spring cloud stream kafka.
Я пробовал так :
spring.cloud.stream.kafka.binder.requiredAcks=all
и
spring.cloud.stream.kafka.streams.binder.configuration=all
и
spring.cloud.stream.kafka.streams.bindings.<channel>.producer.configuration.requiredAcks=all
К сожалению, замечание работает для меня.
Не могли бы вы, пожалуйста, помочь мне, как установить такого рода свойства на уровне приложения или производителя / потребителя.
Комментарии:
1. На всякий случай, используете ли вы Kafka binder или Kafka Streams binder?
Ответ №1:
Приведенная ниже конфигурация используется только связующими средствами Kafka (не kafka Streams). Он используется для установки свойства acks экземпляра производителя.
spring.cloud.stream.kafka.binder.requiredAcks
Чтобы настроить экземпляр Kafka Streams, перед свойствами должен быть префикс spring.cloud.stream.kafka.streams.binder (Конфигурация потока Spring Cloud).
В KafkaStreams свойства producers можно переопределить, добавив префикс «.producer» (см. Настройка приложения Streams; Поэтому для настройки acks producer вам следует определить следующее свойство :
spring.cloud.stream.kafka.streams.binder.configuration.producer.acks=all
Обратите внимание, что если вы создаете приложение Kafka Streams с отслеживанием состояния, настоятельно рекомендуется включить семантику exactly_once.
Эта семантика может быть настроена с помощью :
spring.cloud.stream.kafka.streams.binder.configuration.processing.guarantee=exactly-once
Комментарии:
1. Спасибо за ответ, я взял пример из демо-версии Джоша Лонга. Вот ссылка на файл свойств: [ссылка] github.com/joshlong/spring-cloud-stream-kafka-streams/blob / … , здесь у нас было два производителя, и я хочу настроить свойства для каждого производителя. Как я могу добавить любую требуемую конфигурацию для конкретного производителя, потребителя или уровня приложения?
Ответ №2:
spring.cloud.stream.kafka.binder.configuration.acks: all