{"id":56186,"date":"2020-02-06T00:00:00","date_gmt":"2020-02-05T21:00:00","guid":{"rendered":"https:\/\/prohoster.info\/blog\/blog_prohoster\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka"},"modified":"2020-02-18T14:04:24","modified_gmt":"2020-02-18T11:04:24","slug":"povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","status":"publish","type":"post","link":"https:\/\/prohoster.info\/et\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Kafa \u00fcrituste t\u00f6\u00f6tlemine","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Kafa<\/p>\n

<\/p>\n

Tere, Habr.<\/p>\n

<\/p>\n

Hiljuti jagasin ma oma kogemusi<\/a><\/noindex> selle kohta, milliseid parameetreid me tihti kasutame Kafka producentide ja tarbijate jaoks, et tagada garanteeritud kohaletoimetamine. Selles artiklis r\u00e4\u00e4gin, kuidas korraldasime s\u00fcndmuse taast\u00f6\u00f6tluse, mis saadi Kafka'lt v\u00e4lise s\u00fcsteemi ajutise k\u00e4ttesaamatuse t\u00f5ttu.<\/p>\n

<\/p>\n

Kaasaegsed rakendused toimivad v\u00e4ga keerulises keskkonnas. \u00c4ri\u00fcksus, mis on pakitud kaasaegsesse tehnoloogilisse virna, t\u00f6\u00f6tades Docker'i pildis, mida haldab orkestraator nagu Kubernetes v\u00f5i OpenShift, ja kommunikeerides teiste rakendustega v\u00f5i ettev\u00f5tte lahendustega f\u00fc\u00fcsiliste ja virtuaalsete ruuterite ahela kaudu. Sellises keskkonnas v\u00f5ib alati midagi katki minna, seega on s\u00fcndmuste taast\u00f6\u00f6tlus, kui \u00fcks v\u00e4lisest s\u00fcsteemist ei ole saadaval \u2014 oluline osa meie \u00e4riprotsessidest.<\/p>\n

<\/a><\/noindex><\/p>\n

Kuidas oli enne Kafka<\/h2>\n

<\/p>\n

Varasemalt kasutasime projektis IBM MQ as\u00fcnkroonsete s\u00f5numite saatmiseks. Kui teenuse t\u00f6\u00f6tamise k\u00e4igus tekkis m\u00f5ni viga, v\u00f5is saadud s\u00f5num paigutada dead-letter-queue (DLQ), et seda hiljem k\u00e4sitsi anal\u00fc\u00fcsida. DLQ loodi koos sissetuleva j\u00e4rjekorraga, s\u00f5numi \u00fcmberpaigutamine toimus IBM MQ sees. <\/p>\n

<\/p>\n

Kui viga oli ajutine ja me suutsime selle m\u00e4\u00e4rata (n\u00e4iteks ResourceAccessException HTTP-k\u00f5ne ajal v\u00f5i MongoTimeoutException MongoDb p\u00e4ringu puhul), siis rakendati kordusv\u00e4ljakutsete strateegiat. \u00dcksk\u00f5ik millest rakenduse loogika branch'ist, algne s\u00f5num omistati kas s\u00fcsteemij\u00e4rjekorda edasil\u00fckkamiseks v\u00f5i eraldi rakendusse, mis kunagi ammu loodi s\u00f5numite kordus saatmiseks. S\u00f5numi p\u00e4isesse kirjutatakse kordusv\u00e4ljakutsete number, mis on seotud viivituse intervalliga v\u00f5i rakenduse tasemel strateegia l\u00f5puga. Kui me j\u00f5uame strateegia l\u00f5puni, kuid v\u00e4line s\u00fcsteem on endiselt k\u00e4ttesaamatu, paigutatakse s\u00f5num DLQ-sse k\u00e4sitsi anal\u00fc\u00fcsimiseks.<\/p>\n

<\/p>\n

Lahenduse otsimine<\/h2>\n

<\/p>\n

Otsides internetist<\/a><\/noindex>, v\u00f5ib leida j\u00e4rgmist lahendus<\/a><\/noindex>. L\u00fchidalt \u00f6eldes pakutakse iga viivituse intervalli jaoks luua teema ja rakenduse k\u00fcljes tarbijaid, mis loevad s\u00f5numid vajaliku viivitusega. <\/p>\n

<\/p>\n

\"Kafa<\/p>\n

<\/p>\n

Malgr\u00e9 le grand nombre d'avis positifs, cela ne me semble pas tout \u00e0 fait r\u00e9ussi. Tout d'abord, parce que le d\u00e9veloppeur, en plus de mettre en \u0153uvre les exigences commerciales, devra passer beaucoup de temps \u00e0 mettre en \u0153uvre le m\u00e9canisme d\u00e9crit.<\/p>\n

<\/p>\n

Lisaks, kui Kafka klastri juures on lubatud juurdep\u00e4\u00e4su haldamine, tuleb kulutada aega teemade loomisele ja nendele vajalike juurdep\u00e4\u00e4sude tagamisele. Peale selle on vajalik valida \u00f5ige retention.ms parameeter iga retry-teema jaoks, et s\u00f5numid j\u00f5uaksid uuesti edastada ja ei kaoks. Rakenduse loomine ja juurdep\u00e4\u00e4sude taotlemine tuleb korrata iga olemasoleva v\u00f5i uue teenuse jaoks.<\/p>\n

<\/p>\n

N\u00fc\u00fcd vaatame, milliseid s\u00f5numi uuesti t\u00f6\u00f6tlemise mehhanisme pakub meile spring \u00fcldiselt ja spring-kafka konkreetsemalt. Spring-kafka-l on \u00fclekantav s\u00f5ltuvus spring-retry-st, mis pakub abstraktsioone erinevate BackOffPolicy-de haldamiseks. See on \u00fcsna paindlik t\u00f6\u00f6riist, kuid selle oluline puudus on s\u00f5numite hoidmine uuesti saatmiseks rakenduse m\u00e4lus. See t\u00e4hendab, et rakenduse taask\u00e4ivitamine uuenduse v\u00f5i t\u00f6\u00f6ea vigade t\u00f5ttu toob kaasa k\u00f5ikide uuesti t\u00f6\u00f6tlemist ootavate s\u00f5numite kaotuse. Kuna see on meie s\u00fcsteemi jaoks kriitiline punkt, ei uurinud me seda edaspidi.<\/p>\n

<\/p>\n

Ise spring-kafka pakub mitmeid rakendusi ContainerAwareErrorHandler-le, n\u00e4iteks SeekToCurrentErrorHandler<\/a><\/noindex>, mille abil saab s\u00f5numit hiljem t\u00f6\u00f6delda, mitte nihutades offset'i vea tekkimise korral. Alates versioonist spring-kafka 2.3 on v\u00f5imalus m\u00e4\u00e4rata BackOffPolicy.<\/p>\n

<\/p>\n

See l\u00e4henemine v\u00f5imaldab korduvkasutatavatel s\u00f5numitel rakenduse taask\u00e4ivitamisest \u00fcle 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\u00f5epoolest ei olnud vajalik mitme kuu jooksul rakenduse sellise kordusprotsessimise s\u00fcsteemiga t\u00f6\u00f6tamise ajal). Ajutised vead viisid SeekToCurrentErrorHandleri aktiveerimiseni. \u00dclej\u00e4\u00e4nud vead kanti logisse, mis viis offset'i nihkeni ning t\u00f6\u00f6tlemine j\u00e4tkus j\u00e4rgmise s\u00f5numiga.<\/p>\n

<\/p>\n

L\u00f5plik otsus<\/h2>\n

<\/p>\n

SeekToCurrentErrorHandleril p\u00f5hinev rakendus viis meid oma s\u00f5numite kordussaata mehhanismi v\u00e4ljat\u00f6\u00f6tamiseni.<\/p>\n

<\/p>\n

Esiteks soovisime kasutada juba olemasolevat kogemust ja laiendada seda rakenduse loogika p\u00f5hjal. Rakenduse puhul, millel on lineaarne loogika, oleks optimaalne l\u00f5petada uute s\u00f5numite lugemine l\u00fchikese aja jooksul, mis on m\u00e4\u00e4ratud uuestis kutsumise strateegia raames. Muude rakenduste puhul soovime, et oleks olemas \u00fchtne punkt, mis tagab uuestis kutsumise strateegia rakendamise. Lisaks peaks see \u00fchtne punkt omama DLQ funktsionaalsust m\u00f5lema l\u00e4henemise puhul.<\/p>\n

<\/p>\n

Uuestis kutsumise strateegia peaks olema salvestatud rakenduses, mis vastutab j\u00e4rgmise intervalli saamise eest ajutise vea ilmnemisel.<\/p>\n

<\/p>\n

Consumer'i peatamine lineaarse loogikaga rakendustes<\/h3>\n

<\/p>\n

Spring-kafka kasutamisel v\u00f5ib Consumer'i peatamise kood v\u00e4lja n\u00e4ha umbes nii:<\/p>\n

<\/p>\n

public void pauseListenerContainer(MessageListenerContainer listenerContainer, \n                                   Instant retryAt) {\n        if (nonNull(retryAt) && listenerContainer.isRunning()) {\n            listenerContainer.stop();\n            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);\n            return;\n        }\n        \/\/ DLQ jaoks\n    }<\/code><\/pre>\n
<\/p>\n
N\u00e4ites retryAt on aeg, millal tuleb uuesti k\u00e4ivitada MessageListenerContainer, kui see veel t\u00f6\u00f6tab. Uuesti k\u00e4ivitamine toimub eraldi teemas, mis on k\u00e4ivitatud TaskScheduler'is, mille rakendust pakub ka spring. <\/p>\n
<\/p>\n
M\u00e4\u00e4rame retryAt v\u00e4\u00e4rtuse j\u00e4rgmisel viisil:<\/p>\n
<\/p>\n
    \n
  1. Otsitakse korduvate kutsete loendi v\u00e4\u00e4rtust.<\/li>\n
  2. Loendi v\u00e4\u00e4rtuse p\u00f5hjal otsitakse praegune viivituse intervall korduvate kutsete strateegias. Strateegia kuulutatakse v\u00e4lja rakenduses endas, selle salvestamiseks valisime JSON formaadi.<\/li>\n
  3. JSON-massis leitud intervall sisaldab sekundeid, mille m\u00f6\u00f6dumisel tuleb t\u00f6\u00f6tlemine uuesti k\u00e4ivitada. See sekundite arv liidetakse praegusele ajale, luues v\u00e4\u00e4rtuse retryAt jaoks.<\/li>\n
  4. Kui intervalli ei leita, siis on retryAt v\u00e4\u00e4rtus null ja s\u00f5num saadetakse DLQ-sse k\u00e4sitsi lahendamiseks.<\/li>\n<\/ol>\n
    <\/p>\n
    Sellise l\u00e4henemise puhul j\u00e4\u00e4b alles vaid s\u00e4ilitada iga t\u00f6\u00f6tlemisel oleva teadete korduvate k\u00f5nede arv, n\u00e4iteks rakenduse m\u00e4lus. Katsete arvu salvestamine m\u00e4llu ei ole selle l\u00e4henemise jaoks kriitiline, kuna lineaarse loogikaga rakendus ei saa tervikuna t\u00f6\u00f6tlemist teostada. Erinevalt spring-retry'ist, rakenduse taask\u00e4ivitamine ei too kaasa k\u00f5igi teadete kadumist korduvaks t\u00f6\u00f6tlemiseks, vaid lihtsalt strateegia taask\u00e4ivitamist. <\/p>\n
    <\/p>\n
    See l\u00e4henemine aitab v\u00e4hendada koormust v\u00e4lisele s\u00fcsteemile, mis v\u00f5ib olla kergesti kergesti \u00fcle koormatud ja seet\u00f5ttu mittesaadav. Teisis\u00f5nu, lisaks korduvatele t\u00f6\u00f6tlemistele oleme saavutanud mustri rakendamise. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    Meie juhul on vea k\u00fcnnis vaid 1, ja et minimeerida s\u00fcsteemi seiskamist ajutise n\u00e4tiv\u00f5rgu katkestuse t\u00f5ttu, kasutame v\u00e4ga granulaarset korduvate k\u00f5nede strateegiat v\u00e4ikeste viibimisega intervallidega. See ei pruugi sobida k\u00f5ikidele kontserni rakendustele, seega tuleb vea k\u00fcnnise ja intervalli suuruse vaheline suhe valida, tuginedes s\u00fcsteemi erip\u00e4radele.<\/p>\n
    <\/p>\n
    Eraldi rakendus, mis k\u00e4sitleb rakendustelt saadud s\u00f5numeid, millel on m\u00e4\u00e4ramatu loogika.<\/h3>\n
    <\/p>\n
    Siin on koodin\u00e4ide, mis saadab s\u00f5numit sellesse rakendusse (Retryer), mis saadab j\u00e4lle teemale DESTINATION, kui j\u00f5uab aega RETRY_AT:<\/p>\n
    <\/p>\n
    \npublic <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, \n                         Instant retryAt, String counter, String groupId, Exception e) {\n        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());\n        List<Header> arrayOfHeaders = \n            new ArrayList<>(Arrays.asList(headers.toArray()));\n        updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);\n        updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);\n        updateHeader(arrayOfHeaders, ORIGINAL_PARTITION, \n                     () -> Integer.toString(record.partition()).getBytes());\n        if (nonNull(retryAt)) {\n            updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);\n            updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);\n            updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);\n        } else {\n            updateHeader(arrayOfHeaders, REASON, \n                         ExceptionUtils.getStackTrace(e)::getBytes);\n            updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);\n        }\n        ProducerRecord<K, V> messageToSend =\n            new ProducerRecord<>(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);\n        kafkaTemplate.send(messageToSend);\n    }<\/code><\/pre>\n
    <\/p>\n
    N\u00e4iteks on n\u00e4ha, et palju teavet edastatakse p\u00e4istes. RETRY_AT v\u00e4\u00e4rtus asub samamoodi nagu tarbija peatamise kordamise mehhanismi puhul. Peale DESTINATION ja RETRY_AT edastame:<\/p>\n
    <\/p>\n