Kafa ürituste töötlemine

Tere, Habr.

Hiljuti jagasin ma oma kogemusi selle kohta, milliseid parameetreid me tihti kasutame Kafka producentide ja tarbijate jaoks, et tagada garanteeritud kohaletoimetamine. Selles artiklis räägin, kuidas korraldasime sündmuse taastöötluse, mis saadi Kafka'lt välise süsteemi ajutise kättesaamatuse tõttu.

Kaasaegsed rakendused toimivad väga keerulises keskkonnas. Äriüksus, mis on pakitud kaasaegsesse tehnoloogilisse virna, töötades Docker'i pildis, mida haldab orkestraator nagu Kubernetes või OpenShift, ja kommunikeerides teiste rakendustega või ettevõtte lahendustega füüsiliste ja virtuaalsete ruuterite ahela kaudu. Sellises keskkonnas võib alati midagi katki minna, seega on sündmuste taastöötlus, kui üks välisest süsteemist ei ole saadaval — oluline osa meie äriprotsessidest.

Kuidas oli enne Kafka

Varasemalt kasutasime projektis IBM MQ asünkroonsete sõnumite saatmiseks. Kui teenuse töötamise käigus tekkis mõni viga, võis saadud sõnum paigutada dead-letter-queue (DLQ), et seda hiljem käsitsi analüüsida. DLQ loodi koos sissetuleva järjekorraga, sõnumi ümberpaigutamine toimus IBM MQ sees.

Kui viga oli ajutine ja me suutsime selle määrata (näiteks ResourceAccessException HTTP-kõne ajal või MongoTimeoutException MongoDb päringu puhul), siis rakendati kordusväljakutsete strateegiat. Ükskõik millest rakenduse loogika branch'ist, algne sõnum omistati kas süsteemijärjekorda edasilükkamiseks või eraldi rakendusse, mis kunagi ammu loodi sõnumite kordus saatmiseks. Sõnumi päisesse kirjutatakse kordusväljakutsete number, mis on seotud viivituse intervalliga või rakenduse tasemel strateegia lõpuga. Kui me jõuame strateegia lõpuni, kuid väline süsteem on endiselt kättesaamatu, paigutatakse sõnum DLQ-sse käsitsi analüüsimiseks.

Lahenduse otsimine

Otsides internetist, võib leida järgmist lahendus. Lühidalt öeldes pakutakse iga viivituse intervalli jaoks luua teema ja rakenduse küljes tarbijaid, mis loevad sõnumid vajaliku viivitusega.

Malgré le grand nombre d'avis positifs, cela ne me semble pas tout à fait réussi. Tout d'abord, parce que le développeur, en plus de mettre en œuvre les exigences commerciales, devra passer beaucoup de temps à mettre en œuvre le mécanisme décrit.

Lisaks, kui Kafka klastri juures on lubatud juurdepääsu haldamine, tuleb kulutada aega teemade loomisele ja nendele vajalike juurdepääsude tagamisele. Peale selle on vajalik valida õige retention.ms parameeter iga retry-teema jaoks, et sõnumid jõuaksid uuesti edastada ja ei kaoks. Rakenduse loomine ja juurdepääsude taotlemine tuleb korrata iga olemasoleva või uue teenuse jaoks.

Nüüd vaatame, milliseid sõnumi uuesti töötlemise mehhanisme pakub meile spring üldiselt ja spring-kafka konkreetsemalt. Spring-kafka-l on ülekantav sõltuvus spring-retry-st, mis pakub abstraktsioone erinevate BackOffPolicy-de haldamiseks. See on üsna paindlik tööriist, kuid selle oluline puudus on sõnumite hoidmine uuesti saatmiseks rakenduse mälus. See tähendab, et rakenduse taaskäivitamine uuenduse või tööea vigade tõttu toob kaasa kõikide uuesti töötlemist ootavate sõnumite kaotuse. Kuna see on meie süsteemi jaoks kriitiline punkt, ei uurinud me seda edaspidi.

Ise spring-kafka pakub mitmeid rakendusi ContainerAwareErrorHandler-le, näiteks SeekToCurrentErrorHandler, mille abil saab sõnumit hiljem töödelda, mitte nihutades offset'i vea tekkimise korral. Alates versioonist spring-kafka 2.3 on võimalus määrata BackOffPolicy.

See lähenemine võimaldab korduvkasutatavatel sõnumitel rakenduse taaskäivitamisest üle elada, kuid DLQ mehhanism on endiselt puudu. Just selle valiku tegime 2019. aasta alguses, optimistlikult arvates, et DLQ-d ei ole vaja (meil vedas ja see tõepoolest ei olnud vajalik mitme kuu jooksul rakenduse sellise kordusprotsessimise süsteemiga töötamise ajal). Ajutised vead viisid SeekToCurrentErrorHandleri aktiveerimiseni. Ülejäänud vead kanti logisse, mis viis offset'i nihkeni ning töötlemine jätkus järgmise sõnumiga.

Lõplik otsus

SeekToCurrentErrorHandleril põhinev rakendus viis meid oma sõnumite kordussaata mehhanismi väljatöötamiseni.

Esiteks soovisime kasutada juba olemasolevat kogemust ja laiendada seda rakenduse loogika põhjal. Rakenduse puhul, millel on lineaarne loogika, oleks optimaalne lõpetada uute sõnumite lugemine lühikese aja jooksul, mis on määratud uuestis kutsumise strateegia raames. Muude rakenduste puhul soovime, et oleks olemas ühtne punkt, mis tagab uuestis kutsumise strateegia rakendamise. Lisaks peaks see ühtne punkt omama DLQ funktsionaalsust mõlema lähenemise puhul.

Uuestis kutsumise strateegia peaks olema salvestatud rakenduses, mis vastutab järgmise intervalli saamise eest ajutise vea ilmnemisel.

Consumer'i peatamine lineaarse loogikaga rakendustes

Spring-kafka kasutamisel võib Consumer'i peatamise kood välja näha umbes nii:

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

Näites retryAt on aeg, millal tuleb uuesti käivitada MessageListenerContainer, kui see veel töötab. Uuesti käivitamine toimub eraldi teemas, mis on käivitatud TaskScheduler'is, mille rakendust pakub ka spring.

Määrame retryAt väärtuse järgmisel viisil:

1. Otsitakse korduvate kutsete loendi väärtust.
2. Loendi väärtuse põhjal otsitakse praegune viivituse intervall korduvate kutsete strateegias. Strateegia kuulutatakse välja rakenduses endas, selle salvestamiseks valisime JSON formaadi.
3. JSON-massis leitud intervall sisaldab sekundeid, mille möödumisel tuleb töötlemine uuesti käivitada. See sekundite arv liidetakse praegusele ajale, luues väärtuse retryAt jaoks.
4. Kui intervalli ei leita, siis on retryAt väärtus null ja sõnum saadetakse DLQ-sse käsitsi lahendamiseks.

Sellise lähenemise puhul jääb alles vaid säilitada iga töötlemisel oleva teadete korduvate kõnede arv, näiteks rakenduse mälus. Katsete arvu salvestamine mällu ei ole selle lähenemise jaoks kriitiline, kuna lineaarse loogikaga rakendus ei saa tervikuna töötlemist teostada. Erinevalt spring-retry'ist, rakenduse taaskäivitamine ei too kaasa kõigi teadete kadumist korduvaks töötlemiseks, vaid lihtsalt strateegia taaskäivitamist.

See lähenemine aitab vähendada koormust välisele süsteemile, mis võib olla kergesti kergesti üle koormatud ja seetõttu mittesaadav. Teisisõnu, lisaks korduvatele töötlemistele oleme saavutanud mustri rakendamise. circuit breaker.

Meie juhul on vea künnis vaid 1, ja et minimeerida süsteemi seiskamist ajutise nätivõrgu katkestuse tõttu, kasutame väga granulaarset korduvate kõnede strateegiat väikeste viibimisega intervallidega. See ei pruugi sobida kõikidele kontserni rakendustele, seega tuleb vea künnise ja intervalli suuruse vaheline suhe valida, tuginedes süsteemi eripäradele.

Eraldi rakendus, mis käsitleb rakendustelt saadud sõnumeid, millel on määramatu loogika.

Siin on koodinäide, mis saadab sõnumit sellesse rakendusse (Retryer), mis saadab jälle teemale DESTINATION, kui jõuab aega 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);
    }

Näiteks on näha, et palju teavet edastatakse päistes. RETRY_AT väärtus asub samamoodi nagu tarbija peatamise kordamise mehhanismi puhul. Peale DESTINATION ja RETRY_AT edastame:

• GROUP_ID, mille alusel grupeerime sõnumeid käsitsi analüüsimiseks ja otsingu lihtsustamiseks.
• ORIGINAL_PARTITION, et püüda säilitada sama tarbija uuesti töötlemiseks. See parameeter võib olla null, sel juhul saadakse uus partitsioon originaalsõnumi record.key() võtme järgi.
• Üksuste uuendatud COUNTER väärtus, et järgida korduvate kutsete strateegiat.
• SEND_TO — konstant, mis näitab, kas sõnum tuleks saata uuesti töötlemiseks pärast RETRY_AT saavutamist või paigutada DLQ-sse.
• REASON — põhjus, miks sõnumi töötlemine katkestati.

Retryer salvestab sõnumid edasise saatmise ja käsitsi analüüsimise jaoks PostgreSQL-is. Ajastatult käivitatakse ülesanne, mis leiab sõnumid, mille RETRY_AT on möödunud, ja saadab need tagasi DESTINATIONi TOPICALi ORIGINAL_PARTITION partitsioonile record.key() võtmega.

Pärast sõnumi saatmist kustutatakse need PostgreSQL-ist. Sõnumite käsitsi töötlemine toimub lihtsas kasutajaliideses, mis suheldes Retryeriga üle REST API. Peamised funktsioonid hõlmavad sõnumite uuesti saatmist või DLQ-st kustutamist, veateabe vaatamist ja sõnumite otsimist, näiteks vea nime järgi.

Kuna meie klasterites on lubatud juurdepääsuhaldus, on vaja täiendavalt taotleda juurdepääse teemale, mida kuulab Retryer, ja anda Retryerile võimalus kirjutada DESTINATION teemale. See on ebamugav, kuid võrreldes ajavahemikute põhise lähenemisega on meil täielik DLQ ja kasutajaliides selle haldamiseks.

On juhtumeid, kui sisendteemat loevad mitmed erinevad tarbijagruppide rakendused, millel on erinev loogika. Sõnumi uuesti töötlemine Retryeri kaudu ühe sellise rakenduse jaoks toob teises esile duplikaadi. Selle vastu kaitsmiseks loome eraldi teema uuesti töötlemiseks. Sisend- ja retry-teemat võib lugeda sama tarbija ilma piiranguteta.

Vaikimisi ei paku see lähenemine circuit breaker’i võimalust, kuid selle saab rakendusele lisada spring-cloud-netflix või uus spring cloud ringmurdja, mähkides väliste teenuste kutsumise kohad vastavatesse abstraktsioonidesse. Lisaks ilmub võimalus valida strateegia bulkhead mustrile, mis võib samuti olla kasulik. Näiteks spring-cloud-netflixis võib see olla teepool või seeder.

Kokkuvõte

Tulemuseks on meil eraldi rakendus, mis võimaldab sõnumi töötlemist korrata, kui mõni väline süsteem on ajutiselt kättesaamatu.

Ere rakenduse peamine eelis on see, et seda saavad kasutada välised süsteemid, mis töötavad samas Kafka klastris, ilma märkimisväärsete muudatusteta oma küljel! Selline rakendus peab ainult saama juurdepääsu retry-teemadele, täitma mõned Kafka pealkirjad ja saatma sõnumi Retryerisse. Ühtegi täiendavat infrastruktuuri ei pea üles tõstma. Ja et vähendada sõnumite saadetavaid koguseid rakendusest Retryerisse ja tagasi, oleme eraldanud rakendused, millel on lineaarne loogika, ning teinud neisse uuesti töötlemise kautta Consumeri peatumise.

Allikas: habr.com