Комментарии 15
как это без транзакций в вашем случае? код же такого вида у вас получается
let m = kafka.GetMessage()
DoIncrementFieldInDatabase(m);
// <- электричество выключается тут
kafka.Commit
Тоесть вы допустим энкрементите , а перед комитов падает сервер. В след раз опять поступает это сообщение и базе мы уже получает не валидные данные .
Если я правильно понимаю, то вы описываете такую ситуацию:
Начата транзакция Kafka
Начата транзакция БД
Вычитали сообщение из кафки
Обновили данные в БД
Коммит транзакции БД
<<выключилось электричество>>
<<сообщение из кафки не вычитано (т.е. оффсет не закомитчен) >>
<<после включения электричества то же самое сообщение попадет на повторную обработку>>
Если так, то да, система останется в неконсистентном состоянии, т.к. данные в БД уже обновлены, а то же самое сообщение в кафке пойдет на повторную обработку. Именно про это я и пишу и именно эту ситуацию а моделирую через параметр receive-transactions-faults-num (задайте его > 0 и увидите этот эффект.)
Забегая вперед, скажу, что лечить эту конкретную проблему можно например таким образом: запоминать в БД ключи уже обработанных сообщений и отбрасывать их при повторном получении - паттерн Idempotent Consumer (https://microservices.io/patterns/communication-style/idempotent-consumer.html). Но это усложнение решения и это точно out of scope данной статьи. Если все сложится хорошо - опишем это в отдельной статье.
Если я не ошибаюсь, решение этой проблемы это хранить оффсеты на стороне консюмера. Что то вроде Outbox Pattern только на входящие сообщения.
Kafka Consumer starts
Load the last offset from the db
Receive message from kafka
tr = Start db transaction
handleMessage(tr)
persistOffset(tr)
tr.commit()
Или все или ничего...
Хранить consumer offsets в потребителе (и управлять ими в своем прикладном коде) - это значит брать на себя реализацию существенного объема функционала, который уже реализован в Кафке. Амбициозная задача.
Странный какой-то вы кейс выбрали, возможно специфичный только для BPM. В подавляющем большинстве нужно: прочитать из одного топика, обработать/записать в базу и отправить в другой топик.
Хотелось бы также знать политику консьюмера при роллбеке транзакции: делается ли повтор, как контролировать delay и limit при повторе, отфутболится ли месседж в DLQ, остановить ли весь консьюмер и т.д. Тут масса нюансов и вопросов как это контролировать в спринге.
Вообще, при планировании распределенной архитектуры лучше закладываться на гарантию "at least once" и вручную везде обеспечивать идемпотентность. Насколько я понял, кафка не обеспечивает "exactly once" между двумя топиками.
Почему же странный ? "Получить сообщение + обновить БД" - это элементарный (но при этом вполне реальный) сценарий, на котором проще всего исследовать проблему. И он конечно не специфичен для BPM.
"Прочитать + обновить БД + отправить" - это просто чуть более сложный сценарий, и при этом можно усложнять и далее, комбинируя всё больше взаимодействий - но это вряд ли что-то добавит к сути статьи, скорее наоборот - замаскирует её за второстепенными деталями. То же самое на мой взгляд относится и к retries и dead letters - поддержка всего этого есть в Spring for Apache Kafka, но это выходит за рамки статьи, это вопросы, достойные отдельного рассмотрения. Тем более достойна отдельного рассмотрения тема с "остановкой consumer'а", т.к. тут возникнет куча вопросов с ребалансировкой, порядком обработки сообщений и т.д.
В целом, я считаю, в небольшой статье лучше сфокусироваться на одном вопросе и рассмотреть какой-то простейший (но живой) случай, чтобы понять принцип, и уже дальше этот принцип (с соответствующими расширениями и дополнениями, конечно) использовать в более сложных сценариях.
"Прочитать + обновить БД + отправить"
А какой вообще подход тут может быть? Вроде, видится, что последняя отправка в Кафку должна быть одним из шагов в фиксации транзакции (видимо, первым)? Или можно положиться на менеджер транзакций, который зафиксирует отправку вместе с фиксацией чтения (потому что чисто внутри себя Кафка умеет в сложные транзакции)?
Скорее всего - да, второй вариант: можно положиться на менеджер транзакций Кафки, т.к. внутри Кафки есть ACID-гарантии при работе с несколькими топиками.
потому что чисто внутри себя Кафка умеет в сложные транзакции
Вот фиг его знает. Пишут, что умеет, если выставлен read_committed уровень изоляции.
The consumer's position is stored as a message in a topic, so we can write the offset to Kafka in the same transaction as the output topics receiving the processed data. If the transaction is aborted, the consumer's position will revert to its old value and the produced data on the output topics will not be visible to other consumers, depending on their "isolation level." In the default "read_uncommitted" isolation level, all messages are visible to consumers even if they were part of an aborted transaction, but in "read_committed," the consumer will only return messages from transactions which were committed (and any messages which were not part of a transaction).
Но на практике нигде нет внятной инфы как это физически реализовано (топики-то разные!) Поэтому закладываться на такие гарантии может быть чревато проблемами.
Хорошая тема для исследования и статьи на Хабре )
Что касается деталей реализации, то есть большой дизайн-документ на эту тему - "Exactly Once Delivery and Transactional Messaging in Kafka", https://docs.google.com/document/d/11Jqy_GjUGtdXJK94XGsEIK7CP1SnQGdp2eF0wSw9ra8/edit#heading=h.xq0ee1vnpz4o
>> Очевидно, 1 phase commit best effort - это вероятностный, а не гарантированный метод.
не совсем понятно зачем нужен описанный в статье подход, если можно использовать старый добрый transactional outbox + идемпотентность и получить гарантированный результат.
Удивительно, но я не нашёл в своё время готовой реализации to outbox в spring, но это не трудно реализовать самому.
допускаю, что есть класс задач где best effort достаточен - но тогда было бы интересно обсудить что это за класс задач.
Хороший вопрос )
Пользу описанного здесь подхода я вкратце вижу так: получить хорошую защиту от неконсистентности за небольшую плату.
Плата состоит в том, что вы настраиваете transaction manager'ы и развешиваете @Transactional - это несложно. А в результате получаете, что ошибки в прикладной логике не ломают консистентность вашего приложения - а это основная доля ошибок/сбоев в приложениях. Остается непокрытым риск системных проблем - электричество выключилось, сеть отпала и т.д., но это маловероятные риски, тем более, что для поломки консистентности они должны произойти в очень маленький промежуток времени между коммитами транзакции БД и Кафки.
Суммарно, допустим, вы получаете 99,99% гарантии консистентности при небольших трудозатратах - очень неплохо, особенно по сравнению с полным отсутствием контроля )
Transactional outbox + idempotent consumer - это более существенные трудозатраты, хотя конечно не rocket science. Про eventual consistency планирую сделать отдельную статью.
Распределённые транзакции Kafka + PostgreSQL средствами Spring