#spring-boot #apache-kafka #spring-kafka
Вопрос:
В моем Java-приложении я подключаюсь к 2 разным кластерам Кафки (скажем, C1 и C2), В настоящее время я публикую одно и то же сообщение для обоих этих кластеров(C1 и C2), как показано ниже:
firstKafkaTemplate.send(producerRecord);
secondKafkaTemplate.send(producerRecord);
Я хочу опубликовать сообщение каким-то транзакционным способом, чтобы при публикации сообщения в C1, затем, если
публикация не удалась в C2, сообщение также не должно публиковаться в C1, его следует каким-то образом откатить? Можем ли мы добиться такого поведения?
Комментарии:
1. Добавить потребителя, который читает сообщение в течение определенного периода времени?
2. как это поможет?
3. Если потребитель не прочитает сообщение, оно не будет доставлено в тему. Но перечитывая вопрос, похоже, что вы хотите выполнить запись примерно в то же время и хотите удалить сообщение, если что-то из этого не получится.
4. Если вы используете два кластера, этот тип «охватывающей транзакции» невозможен. Записи Кафки не могут быть откатаны после создания и подтверждения. Вы могли бы встроить логику для своих потребителей, чтобы искать смещения извне и пропускать любые события, соответствующие некоторым условиям, хотя
5. так как у меня нет контроля над потребительскими приложениями. вот почему я искал что-то, что я мог бы сделать со своей стороны, если есть способ отменить отправку сообщения, прервать транзакцию, чтобы сообщение не было опубликовано
Ответ №1:
Кафка не поддерживает распределенные транзакции (так называемые транзакции XA), поэтому вы не можете просто записать данные в оба кластера в одной большой транзакции и иметь возможность откатить ее.
Если вы вообще не хотите менять своих потребителей, вы можете рассмотреть возможность записи только в первый кластер и использовать MirrorMaker или аналогичный инструмент для копирования сообщений из одной темы в другую. Если второму кластеру не удастся записать сообщение, MirrorMaker повторит попытку до тех пор, пока сообщение не будет записано.
Вы также можете рассмотреть возможность мониторинга второго кластера и отключения отправки сообщений в первый кластер, если второй не может получать сообщения.