Здравей, Хабр.
Наскоро споделих опит Съвременните приложения работят в много сложна среда. Бизнес логиката, обвита в съвременен технологичен стек, работеща в Docker образ, управляван от оркестратор като Kubernetes или OpenShift, и комуникираща с други приложения или корпоративни решения чрез верига от физически и виртуални маршрутизатори. В такава среда винаги нещо може да се счупи, затова повторната обработка на събития в случай на недостъпност на една от външните системи е важна част от нашите бизнес процеси.
Как беше преди Kafka
По-рано в проекта използвахме IBM MQ за асинхронно доставяне на съобщения. При възникване на каквато и да е грешка в процеса на работа на услугата, полученото съобщение можеше да бъде поставено в dead-letter-queue (DLQ) за по-нататъшно ръчно разглеждане. DLQ се създаваше до входната опашка, а прехвърлянето на съобщението се извършваше вътре в IBM MQ.
Ако грешката имаше временен характер и можем да го установим (например ResourceAccessException при HTTP повикване или MongoTimeoutException при заявка в MongoDb), то в сила влизаше стратегията за повторни повиквания. Независимо от разклонението на логиката на приложението, оригиналното съобщение се прехвърляше в системна опашка за отложено изпращане или в отделно приложение, което отдавна е направено за повторно изпращане на съобщения. При това в заглавието на съобщението се записва номер на повторно изпращане, свързан с интервала на забавяне или с края на стратегията на ниво приложение. Ако достигнем края на стратегията, но външната система все още е недостъпна, съобщението ще бъде поставено в DLQ за ръчно разглеждане.
Търсене на решение
След търсене в интернет
. Накратко, предлага се да се създаде по тема за всяка интервал на забавяне и да се реализират от страна на приложението Consumer-и, които да прочитат съобщенията с необходимото забавяне. . Кратко казано, предлага се да се създаде тема за всеки интервал на забавяне и да се реализират на страната на приложението Consumer-и, които ще извличат съобщения с необходимото забавяне.
Въпреки множеството положителни отзиви, ми се струва, че не е съвсем успешен. Първо, защото на разработчика, освен да изпълни бизнес изискванията, ще му отнеме много време да реализира описания механизъм.
Освен това, ако в Kafka клъстера е включено управление на достъпа, ще е необходимо да се отдели време за създаване на теми и осигуряване на необходимите права за тях. В допълнение, ще трябва да се подбере правилният параметър retention.ms за всяка от темите за повторни опити, за да могат съобщенията да бъдат повторно изпратени и да не изчезват. Реализацията и заявката за достъп ще трябва да се повтарят за всеки съществуващ или нов сервис.
Сега да разгледаме какви механизми за повторна обработка на съобщения предлагат spring като цяло и spring-kafka в частност. Spring-kafka има транзитивна зависимост от spring-retry, който предоставя абстракции за управление на различни BackOffPolicy. Това е доста гъвкав инструмент, но значителен недостатък е, че съобщенията за повторна изпращане се съхраняват в паметта на приложението. Това означава, че рестартиране на приложението поради обновление или грешка по време на експлоатация ще доведе до загуба на всички съобщения, очакващи повторна обработка. Тъй като този пункт е критичен за нашата система, не разгледахме този вариант по-нататък.
Сама по себе си spring-kafka предлага няколко реализации на ContainerAwareErrorHandler, например , с помощта на който може, без да се променя offset в случай на грешка, да се обработи съобщението по-късно. От версия 2.3 на spring-kafka стана възможно да се задава BackOffPolicy.
Този подход позволява на повторно обработваните съобщения да оцеляват при рестарт на приложението, но механизмът DLQ все още отсъства. Именно този вариант избрахме в началото на 2019 година, оптимистично смятайки, че DLQ няма да бъде необходим (късметна за нас и наистина не беше необходима в продължение на няколко месеца експлоатация на приложението с такава система за повторна обработка). Временните грешки предизвикваха сработване на SeekToCurrentErrorHandler. Останалите грешки се печатаха в логовете, предизвикваха смяна на offset и обработката продължаваше с следващото съобщение.
Крайната решение
Реализация, основана на SeekToCurrentErrorHandler, нас подтолкнула к разработке собственного механизма для повторной отправки сообщений.
Прежде всего, мы хотели воспользоваться существующим опытом и адаптировать его в зависимости от логики приложения. Для приложений с линейной логикой оптимальным решением было бы прекратить считывание новых сообщений на короткий промежуток времени, заданный в рамках стратегии повторных вызовов. Для других приложений мы хотели бы иметь централизованную точку, обеспечивающую выполнение стратегии повторных вызовов. Эта точка должна также обладать функциональностью DLQ для обоих подходов.
Стратегия повторных вызовов должна храниться в приложении, ответственном за получение следующего интервала при возникновении временной ошибки.
Остановка Consumer’a для приложения с линейной логикой.
При работе с spring-kafka код для остановки Consumer’a может выглядеть так:
public void pauseListenerContainer(MessageListenerContainer listenerContainer,
Instant retryAt) {
if (nonNull(retryAt) && listenerContainer.isRunning()) {
listenerContainer.stop();
taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
return;
}
// в DLQ
}В примере retryAt — это время, когда необходимо заново запустить MessageListenerContainer, если он еще работает. Повторный запуск будет происходить в отдельном потоке, созданном в TaskScheduler, реализация которого также предоставляется spring.
Значение retryAt мы определяем следующим образом:
- Ищется значение счетчика повторных вызовов.
- В зависимости от значения счетчика мы определяем текущий интервал задержки в стратегии повторных вызовов. Эта стратегия объявляется в самом приложении, для ее хранения мы выбрали формат JSON.
- Найденный в JSON-массиве интервал содержит количество секунд, через которые необходимо повторить обработку. Это количество секунд прибавляется к текущему времени, формируя значение для retryAt.
- Если интервал не найден, значение retryAt будет равно null, и сообщение отправится в DLQ для ручного разбора.
При този подход остава само да се запази броя на повторните опити за всяко съобщение, което в момента е в обработка, например в паметта на приложението. Запазването на брояча на опитите в паметта не е критично за този подход, тъй като приложението с линейна логика не може да извършва обработка в цялост. В отличие от spring-retry, рестартирането на приложението няма да доведе до загуба на всички съобщения за повторна обработка, а просто до рестартиране на стратегията.
Този подход помага да се намали натоварването на външната система, която може да бъде недостъпна поради много голямо натоварване. С други думи, в допълнение към повторната обработка, постигнахме реализация на патерна .
В нашия случай прагът на грешка е само 1, а за да минимизираме времето на неработоспособност на системата поради временно мрежово нарушение, използваме много гранулна стратегия за повторни опити с малки интервали на закъснение. Това може да не е подходящо за всички приложения на групата компании, затова отношението между прага на грешка и размера на интервала трябва да се подбира, базирайки се на особеностите на системата.
Отделно приложение за обработка на съобщения от приложения с недетерминирана логика
Ето пример за код, който изпраща съобщение в такова приложение (Retryer), което ще извърши повторно изпращане в топик DESTINATION при достигане на времето RETRY_AT:
public void retry(ConsumerRecord record, String retryToTopic,
Instant retryAt, String counter, String groupId, Exception e) {
Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
List arrayOfHeaders =
new ArrayList(Arrays.asList(headers.toArray()));
updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
updateHeader(arrayOfHeaders, ORIGINAL_PARTITION,
() -> Integer.toString(record.partition()).getBytes());
if (nonNull(retryAt)) {
updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
} else {
updateHeader(arrayOfHeaders, REASON,
ExceptionUtils.getStackTrace(e)::getBytes);
updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
}
ProducerRecord messageToSend =
new ProducerRecord(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
kafkaTemplate.send(messageToSend);
} От примера се вижда, че много информация се предава в хедерите. Стойността RETRY_AT е също така, както и за механизма за повторение чрез спиране на Consumer’a. Освен DESTINATION и RETRY_AT, предаваме:
- GROUP_ID, по който групираме съобщенията за ръчен анализ и улесняване на търсенето.
- ORIGINAL_PARTITION, за да се опитаме да запазим същия Consumer за повторна обработка. Този параметър може да бъде null; в такъв случай новата партиция ще бъде получена по ключа record.key() на оригиналното съобщение.
- Актуализирана стойност COUNTER, за да следва стратегията за повторни повиквания.
- SEND_TO — константа, която показва дали да изпратим съобщението за повторна обработка при достигане на RETRY_AT или да го поставим в DLQ.
- REASON — причината, поради която обработката на съобщението е била прекъсната.
Retryer съхранява съобщения за повторно изпращане и ръчно разглеждане в PostgreSQL. По таймер се стартира задача, която намира съобщения с настъпило RETRY_AT и ги изпраща обратно в партицията ORIGINAL_PARTITION на темата DESTINATION с ключа record.key().
След изпращането съобщенията се изтриват от PostgreSQL. Ръчното разглеждане на съобщенията се извършва в прост UI, който взаимодействува с Retryer през REST API. Неговите основни характеристики включват повторно изпращане или изтриване на съобщения от DLQ, преглед на информацията за грешки и търсене на съобщения, например по името на грешката.
Тъй като в нашите клъстери е включено управлението на достъпа, е необходимо допълнително да се искат права за достъп до темата, която слуша Retryer, и да се предостави възможност на Retryer да пише в темата DESTINATION. Това е неудобно, но, за разлика от подхода с темата на интервал, получаваме пълноценна DLQ и UI за управлението ѝ.
Има случаи, когато входящата тема се чете от няколко различни consumer-групи, приложенията на които реализират различна логика. Повторната обработка на съобщението чрез Retryer за едно от тези приложения ще доведе до дубликат на друго. За да се предпазим от това, създаваме отделна тема за повторна обработка. Входящата и retry-темата може да бъде четена от един и същ Consumer без никакви ограничения.
По подразбиране този подход не предоставя възможност за circuit breaker, но може да се добави в приложението с помощта на или новият , обгръщайки местата на извиквания на външни услуги в съответните абстракции. Освен това се появява възможност за избор на стратегия за шаблон, което също може да бъде полезно. Например, в spring-cloud-netflix това може да бъде thread pool или семафор.
Извод
В резултат на това получихме отделно приложение, което позволява повторна обработка на съобщения при временно недостъпна външна система.
Едно от основните предимства на приложението е, че външните системи, работещи на същия Kafka-клъстер, могат да го използват без значителни промени от тяхна страна! На такова приложение просто ще му е необходимо да получи достъп до retry-топика, да попълни няколко Kafka-заглавия и да изпрати съобщение в Retryer. Не е необходимо да се изгражда никаква допълнителна инфраструктура. И за да се намали броят на прехвърлените съобщения между приложението и Retryer, отделихме приложенията с линейна логика и направихме повторната обработка чрез спиране на Consumer.
Източник: habr.com
