Sündmuste uuesti töötlemine, mis on saadud Kafka'st

Tere, Habr.

Hiljuti jagasin ma oma kogemust selle kohta, milliseid parameetreid me tiimis kõige sagedamini kasutame Kafka Producentide ja Tarbijate jaoks, et saavutada garanteeritud kohaletoimetamine. Selles artiklis tahan rääkida, kuidas me korraldasime sündmuse uuesti töötlemise, mis saadi Kafka'st, välise süsteemi ajutise kättesaamatuse tõttu.

Kaasaegsed rakendused töötavad väga keerulises keskkonnas. Ärilogika, mis on mähitud kaasaegsesse tehnoloogilisse kihti, töötab Docker'i pildis, mida haldab koordineerija nagu Kubernetes või OpenShift, ning suhtleb teiste rakenduste või ettevõtte lahendustega läbi füüsiliste ja virtuaalsete marsruuterite ahela. Sellises keskkonnas võib alati midagi katki minna, seetõttu on sündmuste uuesti töötlemine juhul, kui mõni väline süsteem on kättesaamatu, meie äri protsesside oluline osa.

Kuidas oli enne Kafka

Varem kasutasime projektis IBM MQ asünkroonseks sõnumite edastamiseks. Teenuse töö käigus tekkinud vea korral võis saadud sõnum paigutada sisemise surnud sõnumite järjekorda (DLQ) edasiseks käsitlemiseks. DLQ loodi siseneva järjekorra kõrvale, sõnumi edasiviimine toimus IBM MQ sees.

Kui viga oli ajutine ja me suudame selle tuvastada (näiteks ResourceAccessException HTTP-kõne käigus või MongoTimeoutException MongoDb päringus), siis rakendati korduste strateegiat. Sõltumata rakenduse loogika harudest, paigutati algne sõnum kas süsteemijärjekorda edasilükatud saatmiseks või eraldi rakendusse, mis kunagi ammu loodi sõnumite uuesti saatmiseks. Samal ajal kirjutatakse sõnumi päisesse saadetiste korduse number, mis on seotud viivituse intervalliga või rakenduse taseme strateegia lõpuga. Kui oleme jõudnud strateegia lõppu, kuid väline süsteem on endiselt kättesaamatu, siis paigutatakse sõnum DLQ-sse käsitlemiseks.

Lahenduse otsing

Otsides internetist, võib leida järgmist otsuse. Lühidalt öeldes pakutakse välja iga teema jaoks luua topikud igaks viivituse intervalliks ja rakendada rakenduse poolel Tarbijad, kes loevad sõnumeid vajaliku viivitusega.

Hoolimata paljusid positiivsetest ülevaadetest, näib see mulle mitte just kõige parem. Esiteks seetõttu, et arendajal tuleb peale ärinõuete rakendamise kulutada palju aega ka kirjeldatud mehhanismi rakendamisele.

Lisaks, kui Kafka-klastris on aktiveeritud juurdepääsude haldamine, peab kulutama aega teemade loomisele ja nendele vajalike juurdepääsude tagamisele. Sellele lisandub vajadus leida iga retry-teema jaoks õige retention.ms parameeter, et sõnumid jõuaksid uuesti saata ja ei kaoks. Juhtimise ja juurdepääsutaotluse rakendamine tuleb korrata iga olemasoleva või uue teenuse jaoks.

Vaatame nüüd, milliseid mehhanisme sõnumite uuesti töötlemiseks pakub meile Spring tervikuna ja spring-kafka eelkõige. Spring-kafka-l on ülekantav sõltuvus spring-retry'st, mis pakub abstraheerimisvõimalusi erinevate BackOffPolicy haldamiseks. See on üsna paindlik tööriist, kuid selle oluline puudus on see, et sõnumid uuesti saatmiseks salvestatakse rakenduse mällu. See tähendab, et rakenduse taaskäivitamine seoses uuendamise või tööea jooksul tekkinud veaga toob kaasa kõikide uuesti töötlemist ootavate sõnumite kadumise. Kuna see punkt on meie süsteemi jaoks kriitiline, ei kaalunud me seda edasi.

Spring-kafka ise pakub mitmeid ContainerAwareErrorHandler'i rakendusi, näiteks SeekToCurrentErrorHandler, millega saab, mitte nihutades offsetit vea tekkimisel, sõnumi hiljem töödelda. Alates versioonist spring-kafka 2.3 on võimalik määrata BackOffPolicy.

See lähenemine võimaldab uuesti töödeldavatel sõnumitel ellu jääda rakenduse taaskäivitamise, kuid DLQ mehhanism on endiselt puudulik. Just selle variandi valisime 2019. aasta alguses, optimistlikult pidades, et DLQ-d ei tule vaja (meil vedas ja tõepoolest ei olnud seda mitu kuud rakenduse sellise uuesti töötlemise süsteemiga). Ajutised vead põhjustasid SeekToCurrentErrorHandler'i aktiveerimist. Ülejäänud vead kirjutatakse logisse, põhjustavad offseti nihkumist ja töötlemine jätkub järgmise sõnumiga.

Lõplik lahendus

SeekToCurrentErrorHandler'ile põhinev rakendus viitis meid arendada oma mehhanismi sõnumite edastamiseks uuesti.

Esiteks soovisime kasutada juba olemasolevat teadmist ja laiendada seda rakenduse loogika põhjal. Lineaarse loogikaga rakenduse jaoks oleks optimaalne peatada uute sõnumite lugemine lühikeseks ajaks, mis on määratud uuesti kõnede strateegia raames. Ülejäänud rakenduste puhul soovisime omada ühte punkti, mis tagaks uuesti kõnede strateegia elluviimise. Lisaks peaks see ühtne punkt omama DLQ funktsionaalsust mõlema lähenemise jaoks.

Uuesti kõnede strateegia peaks olema salvestatud rakenduses, mis vastutab järgmise intervalli saamise eest ajutiste vigade korral.

Consumer'i peatamine lineaarse loogikaga rakenduse jaoks

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

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

Näites on retryAt aeg, millal tuleks uuesti käivitada MessageListenerContainer, kui see veel töötab. Uuesti käivitamine toimub eraldi lõimes, mis on käivitatud TaskScheduler'is, mille rakenduse pakub samuti spring.

Löime retryAt väärtuse järgmise meetodi abil:

1. Otsitakse uuesti kõnede arvu väärtust.
2. Vastavalt uuesti kõnede arvu väärtusele otsitakse praegune viivituse intervall uuesti kõnede strateegias. Strateegia kuulutatakse välja rakenduses, mille salvestamiseks valisime JSON formaadi.
3. JSON-massis leitud intervall sisaldab sekundeid, mille jooksul tuleb töötlemine uuesti korrata. See sekundite arv liidetakse praegusele ajale, moodustades retryAt väärtuse.
4. Kui intervalli ei leita, siis retryAt väärtus on null ja sõnum saadetakse DLQ-sse käsitsi töötlemiseks.

Selle lähenemise puhul on vajalik säilitada igas sõnumis, mis on praegu töötlemisel, korduvate katsete arv, näiteks rakenduse mälu. Katsete arvu säilitamine mälus ei ole kriitiline, kuna lineaarse loogikaga rakendus ei pruugi kogu töötlemist korraldada. Erinevalt spring-retry'st ei too rakenduse taaskäivitamine kaasa kõigi sõnumite kadumist uuesti töötlemiseks, vaid lihtsalt strateegia taaskäivitamise.

See lähenemine aitab vähendada koormust välisele süsteemile, mis võib olla väga suure koormuse tõttu kättesaamatu. Teisisõnu, lisaks uuesti töötlemisele oleme saavutanud mustri elluviimise. circuit breaker.

Meie puhul on veapiirang vaid 1, ja et minimeerida süsteemi seiskumist ajutiste võrgu katkestuste tõttu, kasutame väga granulaarses korduste strateegias lühikesi viivitusintervalle. See võib mitte sobida kõigile ettevõtte grupi rakendustele, seetõttu tuleb veapiirangu ja intervalli suuruse suhet valida, tuginedes süsteemi omadustele.

Eraldi rakendus sõnumite töötlemiseks rakendustelt, millel on määramatud loogikad.

Siin on näide koodist, mis saadab sõnumi sellesse rakendusse (Retryer), mis täidab uuesti saatmise DESTINATION teema juurde, kui RETRY_AT aeg on käes:

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äidatud näitest on näha, et palju teavet edastatakse pealkirjades. RETRY_AT väärtus on samuti sama, mis Consumer’i peatamise kordusmehhanismi puhul. Lisaks DESTINATIONile ja RETRY_AT-ile edastame:

• GRUP_ID, mille kaudu rühmitame sõnumid käsitsi analüüsimiseks ja otsingu lihtsustamiseks.
• ALGE_PARTITSIOON, et püsida sama Consumeri juures ümber töötlemiseks. See parameeter võib olla null, sel juhul omandatakse uus partitsioon originaalsõnumi record.key() võtme järgi.
• Uuendatud COUNTER väärtus, et järgida korduvate kõnede strateegiat.
• SEND_TO — konstant, mis näitab, kas sõnum saata korduvaks töötlemiseks RETRY_AT saavutatuna või paigutada DLQ-sse.
• PÕHJUS — põhjus, miks sõnumi töötlemine katkestati.

Retryer salvestab sõnumeid korduvaks saatmiseks ja käsitsi analüüsimiseks PostgreSQL-is. Ajastatult käivitatakse ülesanne, mis leiab sõnumid, mille RETRY_AT on saabunud, ja saadab need tagasi originaalsesse partitsiooni DESTINATION teemas record.key() võtmega.

Pärast sõnumi saatmist eemaldatakse need PostgreSQL-ist. Sõnumite käsitsi analüüs toimub lihtsas kasutajaliideses, mis suhtleb Retryeriga REST API kaudu. Selle peamised omadused on sõnumite uuesti saatmine või DLQ-st kustutamine, vigade teabe vaatamine ja sõnumite otsimine, näiteks veateate järgi.

Kuna meie klastrites on sisse lülitatud juurdepääsu haldamine, tuleb täiendavalt taotleda juurdepääsu teemale, mida kuulab Retryer, ja anda Retryerile võimalus kirjutada DESTINATION teemas. See on ebamugav, kuid erinevalt ajavahemikus põhinevast lähenemisest on meil täielik DLQ ja UI selle haldamiseks.

On juhtumeid, kus sisenemise teemat loevad erinevad consumer-grupid, mille rakendused rakendavad erinevat loogikat. Sõnumi korduv töötlemine Retryeri kaudu ühe sellise rakenduse jaoks põhjustab teise rakenduse duplikaadi. Selle kaitsmiseks loome eraldi teema korduvaks töötlemiseks. Sisenemis- ja retry-teemat võib lugeda sama Consumeriga piiramatu arv kordi.

Vaikimisi ei paku see lähenemine circuit breaker’i võimalust, kuid seda saab rakendusele lisada spring-cloud-netflix või uue spring cloud circuit breaker, ümbritsedes väliste teenuste kutsumise kohti vastava abstraktsiooniga. Lisaks ilmneb võimalus valida strateegia bulkhead mustri jaoks, mis võib samuti kasulik olla. Näiteks spring-cloud-netflixis võib see olla niidi bassein või semafor.

Kokkuvõte

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

Rakenduse üks peamisi eeliseid on see, et seda saavad kasutada välist süsteemi, mis töötab samas Kafka-klistris, ilma oluliste muudatusteta oma poolel! Selline rakendus peab vaid saama juurdepääsu retry-teemale, täitma mõned Kafka-pealkirjad ja saatma sõnumi Retryerisse. Ei ole vaja luua täiendavat infrastruktuuri. Ja et vähendada sõnumite edasisaatmist rakendusest Retryerisse ja tagasi, oleme eraldanud rakendused, millel on lineaarsed loogikad, ja viime nende kaudu kordustöötlemise läbi Consumeri peatamise.

Allikas: habr.com