Поновна обрада догађаја примљених од Кафке

Хеј Хабр.

Недавно сам ја поделио своје искуство о томе које параметре ми као тим најчешће користимо за Кафка произвођача и потрошача да би се приближили гарантованој испоруци. У овом чланку желим да вам кажем како смо организовали поновну обраду догађаја добијеног од Кафке као резултат привремене недоступности спољног система.

Савремене апликације раде у веома сложеном окружењу. Пословна логика умотана у модерну технологију, која ради у Доцкер имиџу којим управља оркестратор као што је Кубернетес или ОпенСхифт, и комуницира са другим апликацијама или пословним решењима кроз ланац физичких и виртуелних рутера. У таквом окружењу увек се нешто може покварити, тако да је поновна обрада догађаја ако је један од спољних система недоступан важан део наших пословних процеса.

Како је било пре Кафке

Раније у пројекту користили смо ИБМ МК за асинхрону испоруку порука. Ако је дошло до било какве грешке током рада услуге, примљена порука би се могла ставити у ред за мртве поруке (ДЛК) ради даљег ручног рашчлањивања. ДЛК је креиран поред долазног реда, порука је пренета унутар ИБМ МК.

Ако је грешка била привремена и могли бисмо да је утврдимо (на пример, РесоурцеАццессЕкцептион на ХТТП позиву или МонгоТимеоутЕкцептион на МонгоДб захтеву), тада би стратегија поновног покушаја ступила на снагу. Без обзира на логику гранања апликације, оригинална порука је премештена или у системски ред за одложено слање, или у посебну апликацију која је давно направљена за поновно слање порука. Ово укључује број поновног слања у заглављу поруке, који је везан за интервал кашњења или крај стратегије на нивоу апликације. Ако смо дошли до краја стратегије, али спољни систем још увек није доступан, порука ће бити смештена у ДЛК ради ручног рашчлањивања.

Проналажење рјешења

Претраживање на Интернету, можете пронаћи следеће одлука. Укратко, предлаже се креирање теме за сваки интервал кашњења и имплементација потрошачких апликација са стране, које ће читати поруке са потребним кашњењем.

Упркос великом броју позитивних критика, чини ми се да то није сасвим успешно. Пре свега, зато што ће програмер, поред имплементације пословних захтева, морати да потроши доста времена на имплементацију описаног механизма.

Поред тога, ако је контрола приступа омогућена на Кафка кластеру, мораћете да проведете неко време да креирате теме и обезбедите им неопходан приступ. Поред овога, мораћете да изаберете исправан параметар ретентион.мс за сваку тему за поновни покушај како би поруке имале времена да се поново пошаљу и не би нестале из ње. Имплементација и захтев за приступ ће се морати поновити за сваку постојећу или нову услугу.

Хајде сада да видимо које нам механизме спринг уопште и спринг-кафка посебно пружају за поновну обраду порука. Спринг-кафка има транзитивну зависност од спринг-ретри, што обезбеђује апстракције за управљање различитим БацкОффПолициес. Ово је прилично флексибилан алат, али његов значајан недостатак је складиштење порука за поновно слање у меморији апликације. То значи да ће поновно покретање апликације због ажурирања или оперативне грешке довести до губитка свих порука које чекају на поновну обраду. Пошто је ова тачка критична за наш систем, нисмо је даље разматрали.

Спринг-кафка сама обезбеђује неколико имплементација ЦонтаинерАвареЕррорХандлер, на пример СеекТоЦуррентЕррорХандлер, са којим можете касније обрадити поруку без померања офсета у случају грешке. Почевши од верзије спринг-кафка 2.3, постало је могуће подесити БацкОффПолици.

Овај приступ омогућава да поново обрађене поруке преживе поновно покретање апликације, али још увек не постоји ДЛК механизам. Ову опцију смо изабрали почетком 2019. године, оптимистично верујући да ДЛК неће бити потребан (имали смо среће и заправо нам није био потребан након неколико месеци рада апликације са таквим системом за поновну обраду). Привремене грешке су довеле до покретања СеекТоЦуррентЕррорХандлер-а. Преостале грешке су одштампане у дневнику, што је резултирало офсетом, а обрада је настављена са следећом поруком.

Коначна одлука

Имплементација заснована на СеекТоЦуррентЕррорХандлер-у нас је подстакла да развијемо сопствени механизам за поновно слање порука.

Пре свега, желели смо да искористимо постојеће искуство и проширимо га у зависности од логике апликације. За примену линеарне логике, било би оптимално да престанете да читате нове поруке на кратак временски период одређен стратегијом поновног покушаја. За друге апликације, желео сам да имам једну тачку која би применила стратегију поновног покушаја. Поред тога, ова појединачна тачка мора имати ДЛК функционалност за оба приступа.

Сама стратегија поновног покушаја мора бити сачувана у апликацији, која је одговорна за преузимање следећег интервала када дође до привремене грешке.

Заустављање потрошача за апликацију линеарне логике

Када радите са спринг-кафка, код за заустављање потрошача може изгледати отприлике овако:

public void pauseListenerContainer(MessageListenerContainer listenerContainer, 
                                   Instant retryAt) {
        if (nonNull(retryAt) && listenerContainer.isRunning()) {
            listenerContainer.stop();
            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
            return;
        }
        // to DLQ
    }

У примеру, ретриАт је време за поновно покретање МессагеЛистенерЦонтаинер ако је још увек покренут. Поновно покретање ће се десити у посебној нити покренутој у ТаскСцхедулер-у, чија имплементација је такође обезбеђена до пролећа.

Проналазимо ретриАт вредност на следећи начин:

1. Тражи се вредност бројача поновног позива.
2. На основу вредности бројача, претражује се тренутни интервал кашњења у стратегији поновног покушаја. Стратегија је декларисана у самој апликацији, изабрали смо ЈСОН формат за складиштење.
3. Интервал пронађен у ЈСОН низу садржи број секунди након којих ће се обрада морати поновити. Овај број секунди се додаје тренутном времену да би се формирала вредност за ретриАт.
4. Ако интервал није пронађен, онда је вредност ретриАт нулта и порука ће бити послата у ДЛК ради ручног рашчлањивања.

Са овим приступом, остаје само да се сачува број поновљених позива за сваку поруку која се тренутно обрађује, на пример у меморији апликације. Чување броја поновних покушаја у меморији није критично за овај приступ, пошто апликација линеарне логике не може да се носи са обрадом у целини. За разлику од пролећног поновног покушаја, поновно покретање апликације неће проузроковати да се све поруке изгубе поново обрађују, већ ће једноставно поново покренути стратегију.

Овај приступ помаже да се скине оптерећење са спољног система, који може бити недоступан због веома великог оптерећења. Другим речима, поред поновне обраде, постигли смо имплементацију шаблона прекидач.

У нашем случају, праг грешке је само 1, а да бисмо минимизирали застоје система због привремених прекида мреже, користимо веома детаљну стратегију поновног покушаја са малим интервалима кашњења. Ово можда није погодно за све групне апликације, тако да се однос између прага грешке и вредности интервала мора изабрати на основу карактеристика система.

Посебна апликација за обраду порука од апликација са недетерминистичком логиком

Ево примера кода који шаље поруку таквој апликацији (Ретриер), која ће поново послати тему ДЕСТИНАТИОН када се достигне време РЕТРИ_АТ:

public <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, 
                         Instant retryAt, String counter, String groupId, Exception e) {
        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
        List<Header> 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<K, V> messageToSend =
            new ProducerRecord<>(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
        kafkaTemplate.send(messageToSend);
    }

Пример показује да се много информација преноси у заглављима. Вредност РЕТРИ_АТ се налази на исти начин као и за механизам поновног покушаја кроз Цонсумер стоп. Поред ДЕСТИНАТИОН и РЕТРИ_АТ пролазимо:

• ГРОУП_ИД, помоћу које групишемо поруке за ручну анализу и поједностављену претрагу.
• ОРИГИНАЛ_ПАРТИТИОН да покуша да задржи истог потрошача за поновну обраду. Овај параметар може бити нулл, у ком случају ће се нова партиција добити помоћу кључа рецорд.кеи() оригиналне поруке.
• Ажурирана вредност ЦОУНТЕР да би се пратила стратегија поновног покушаја.
• СЕНД_ТО је константа која указује на то да ли се порука шаље на поновну обраду када достигне РЕТРИ_АТ или је смештена у ДЛК.
• РАЗЛОГ - разлог због којег је обрада поруке прекинута.

Ретриер складишти поруке за поновно слање и ручно рашчлањивање у ПостгреСКЛ. Тајмер покреће задатак који проналази поруке са РЕТРИ_АТ и шаље их назад на ОРИГИНАЛ_ПАРТИТИОН партицију теме ДЕСТИНАТИОН помоћу кључа рецорд.кеи().

Када се пошаљу, поруке се бришу из ПостгреСКЛ-а. Ручно рашчлањивање порука се дешава у једноставном корисничком интерфејсу који је у интеракцији са Ретриер-ом преко РЕСТ АПИ-ја. Његове главне карактеристике су поновно слање или брисање порука из ДЛК-а, преглед информација о грешци и тражење порука, на пример по имену грешке.

Пошто је контрола приступа омогућена на нашим кластерима, потребно је додатно затражити приступ теми коју Ретриер слуша и дозволити Ретриеру да пише у тему ОДРЕДИШТЕ. Ово је незгодно, али, за разлику од приступа интервалним темама, имамо пуноправни ДЛК и кориснички интерфејс за управљање.

Постоје случајеви када долазну тему чита неколико различитих група потрошача, чије апликације имплементирају различиту логику. Поновна обрада поруке преко Ретриер-а за једну од ових апликација ће довести до дупликата на другој. Да бисмо се заштитили од овога, креирамо посебну тему за поновну обраду. Долазне теме и теме за поновни покушај може да чита исти потрошач без икаквих ограничења.

Овај приступ подразумевано не обезбеђује функционалност прекидача, али се може додати апликацији помоћу спринг-цлоуд-нетфлик или нови пролећни облачни прекидач, умотавајући места где се екстерне услуге позивају у одговарајуће апстракције. Поред тога, постаје могуће изабрати стратегију за преграда образац, који такође може бити користан. На пример, у спринг-цлоуд-нетфлик-у ово може бити скуп нити или семафор.

Излаз

Као резултат тога, имамо засебну апликацију која нам омогућава да поновимо обраду поруке ако је било који спољни систем привремено недоступан.

Једна од главних предности апликације је то што је могу користити спољни системи који раде на истом Кафка кластеру, без значајних модификација на њиховој страни! Таква апликација ће морати само да приступи теми за поновни покушај, попуни неколико Кафка заглавља и пошаље поруку Ретриер-у. Нема потребе за подизањем додатне инфраструктуре. А да бисмо смањили број порука пренетих из апликације у Ретриер и назад, идентификовали смо апликације са линеарном логиком и поново их обрађивали кроз Цонсумер стоп.

Извор: ввв.хабр.цом