Assert Kafka отправить работал

Question

Я пишу приложение с помощью Spring Boot, поэтому для записи в Kafka я делаю:

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

, а затем в моем методе:

kafkaTemplate.send(topic, data)

Но я чувствую, что я 'Я просто полагаюсь, что это сработает, как я могу узнать, сработало ли это?Если он асинхронный, стоит ли возвращать код 200 и надеяться, что он работает?Я не совсем понимаю.Если Кафка недоступна, не потерпит ли это неудачу?Разве мне не нужно было ловить исключение?

Answer 1

Да, если Kafka недоступен, этот вызов .send() не будет выполнен, но если вы отправите его асинхронно, никто не будет уведомлен.Вы можете указать обратный вызов, который вы хотите выполнить, когда будущее, наконец, закончится.Полная спецификация интерфейса здесь: https://kafka.apache.org/20/javadoc/org/apache/kafka/clients/producer/Callback.html

Из официального Javadoc Kafka здесь: https://kafka.apache.org/20/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html

Полностью неблокирующее использование может использовать параметр Callback для обеспечения обратного вызова, которыйбудет вызван после завершения запроса.

  ProducerRecord<byte[],byte[]> record = new ProducerRecord<byte[],byte[]>("the-topic", key, value); 
      producer.send(myRecord,
           new Callback() {
               public void onCompletion(RecordMetadata metadata, Exception e) {
                   if(e != null) {
                      e.printStackTrace();
                   } else {
                      System.out.println("The offset of the record we just sent is: " + metadata.offset());
                   }
               }
           });

Answer 2

Наряду с тем, что упомянул @mjuarez, вы можете попробовать поиграть с двумя свойствами производителя Kafka.Одним из них является ProducerConfig.ACKS_CONFIG, что позволяет вам установить уровень подтверждения, который, по вашему мнению, безопасен для вашего варианта использования.Эта ручка имеет три возможных значения.От Kafka doc

• acks=0: Производитель не заботится о подтверждении от сервера и считает его отправленным.
• acks=1: Это будетозначает, что руководитель записывает запись в свой локальный журнал, но отвечает, не ожидая полного подтверждения от всех подписчиков.
• acks=all: Это означает, что руководитель будет ожидать полного набора синхронных реплик для подтверждениязапись.

Другое свойство - ProducerConfig.RETRIES_CONFIG.Установка значения больше нуля приведет к повторной отправке клиентом любой записи, отправка которой завершится неудачно с потенциально кратковременной ошибкой.

Answer 3

вы можете использовать приведенную ниже команду при отправке сообщений на kafka:

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic topic-name

во время выполнения вышеуказанной команды вы должны запустить свой код, и если отправка сообщений прошла успешно, то сообщение должно быть напечатано на консоли.

Кроме того, аналогично любому другому соединению с любыми ресурсами, если соединение не может быть установлено, выполнение каких-либо операций приведет к возникновению некоторых исключений.