{"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":"S\u00fcndmuste uuesti t\u00f6\u00f6tlemine, mis on saadud Kafka'st","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"S\u00fcndmuste<\/p>\n

<\/p>\n

Tere, Habr.<\/p>\n

<\/p>\n

Hiljuti jagasin ma oma kogemust<\/a><\/noindex> selle kohta, milliseid parameetreid me tiimis k\u00f5ige sagedamini kasutame Kafka Producentide ja Tarbijate jaoks, et saavutada garanteeritud kohaletoimetamine. Selles artiklis tahan r\u00e4\u00e4kida, kuidas me korraldasime s\u00fcndmuse uuesti t\u00f6\u00f6tlemise, mis saadi Kafka'st, v\u00e4lise s\u00fcsteemi ajutise k\u00e4ttesaamatuse t\u00f5ttu.<\/p>\n

<\/p>\n

Kaasaegsed rakendused t\u00f6\u00f6tavad v\u00e4ga keerulises keskkonnas. \u00c4rilogika, mis on m\u00e4hitud kaasaegsesse tehnoloogilisse kihti, t\u00f6\u00f6tab Docker'i pildis, mida haldab koordineerija nagu Kubernetes v\u00f5i OpenShift, ning suhtleb teiste rakenduste v\u00f5i ettev\u00f5tte lahendustega l\u00e4bi f\u00fc\u00fcsiliste ja virtuaalsete marsruuterite ahela. Sellises keskkonnas v\u00f5ib alati midagi katki minna, seet\u00f5ttu on s\u00fcndmuste uuesti t\u00f6\u00f6tlemine juhul, kui m\u00f5ni v\u00e4line s\u00fcsteem on k\u00e4ttesaamatu, meie \u00e4ri protsesside oluline osa.<\/p>\n

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

Kuidas oli enne Kafka<\/h2>\n

<\/p>\n

Varem kasutasime projektis IBM MQ as\u00fcnkroonseks s\u00f5numite edastamiseks. Teenuse t\u00f6\u00f6 k\u00e4igus tekkinud vea korral v\u00f5is saadud s\u00f5num paigutada sisemise surnud s\u00f5numite j\u00e4rjekorda (DLQ) edasiseks k\u00e4sitlemiseks. DLQ loodi siseneva j\u00e4rjekorra k\u00f5rvale, s\u00f5numi edasiviimine toimus IBM MQ sees. <\/p>\n

<\/p>\n

Kui viga oli ajutine ja me suudame selle tuvastada (n\u00e4iteks ResourceAccessException HTTP-k\u00f5ne k\u00e4igus v\u00f5i MongoTimeoutException MongoDb p\u00e4ringus), siis rakendati korduste strateegiat. S\u00f5ltumata rakenduse loogika harudest, paigutati algne s\u00f5num kas s\u00fcsteemij\u00e4rjekorda edasil\u00fckatud saatmiseks v\u00f5i eraldi rakendusse, mis kunagi ammu loodi s\u00f5numite uuesti saatmiseks. Samal ajal kirjutatakse s\u00f5numi p\u00e4isesse saadetiste korduse number, mis on seotud viivituse intervalliga v\u00f5i rakenduse taseme strateegia l\u00f5puga. Kui oleme j\u00f5udnud strateegia l\u00f5ppu, kuid v\u00e4line s\u00fcsteem on endiselt k\u00e4ttesaamatu, siis paigutatakse s\u00f5num DLQ-sse k\u00e4sitlemiseks.<\/p>\n

<\/p>\n

Lahenduse otsing<\/h2>\n

<\/p>\n

Otsides internetist<\/a><\/noindex>, v\u00f5ib leida j\u00e4rgmist otsuse<\/a><\/noindex>. L\u00fchidalt \u00f6eldes pakutakse v\u00e4lja iga teema jaoks luua topikud igaks viivituse intervalliks ja rakendada rakenduse poolel Tarbijad, kes loevad s\u00f5numeid vajaliku viivitusega. <\/p>\n

<\/p>\n

\"S\u00fcndmuste<\/p>\n

<\/p>\n

Hoolimata paljusid positiivsetest \u00fclevaadetest, n\u00e4ib see mulle mitte just k\u00f5ige parem. Esiteks seet\u00f5ttu, et arendajal tuleb peale \u00e4rin\u00f5uete rakendamise kulutada palju aega ka kirjeldatud mehhanismi rakendamisele.<\/p>\n

<\/p>\n

Lisaks, kui Kafka-klastris on aktiveeritud juurdep\u00e4\u00e4sude haldamine, peab kulutama aega teemade loomisele ja nendele vajalike juurdep\u00e4\u00e4sude tagamisele. Sellele lisandub vajadus leida iga retry-teema jaoks \u00f5ige retention.ms parameeter, et s\u00f5numid j\u00f5uaksid uuesti saata ja ei kaoks. Juhtimise ja juurdep\u00e4\u00e4sutaotluse rakendamine tuleb korrata iga olemasoleva v\u00f5i uue teenuse jaoks.<\/p>\n

<\/p>\n

Vaatame n\u00fc\u00fcd, milliseid mehhanisme s\u00f5numite uuesti t\u00f6\u00f6tlemiseks pakub meile Spring tervikuna ja spring-kafka eelk\u00f5ige. Spring-kafka-l on \u00fclekantav s\u00f5ltuvus spring-retry'st, mis pakub abstraheerimisv\u00f5imalusi erinevate BackOffPolicy haldamiseks. See on \u00fcsna paindlik t\u00f6\u00f6riist, kuid selle oluline puudus on see, et s\u00f5numid uuesti saatmiseks salvestatakse rakenduse m\u00e4llu. See t\u00e4hendab, et rakenduse taask\u00e4ivitamine seoses uuendamise v\u00f5i t\u00f6\u00f6ea jooksul tekkinud veaga toob kaasa k\u00f5ikide uuesti t\u00f6\u00f6tlemist ootavate s\u00f5numite kadumise. Kuna see punkt on meie s\u00fcsteemi jaoks kriitiline, ei kaalunud me seda edasi.<\/p>\n

<\/p>\n

Spring-kafka ise pakub mitmeid ContainerAwareErrorHandler'i rakendusi, n\u00e4iteks SeekToCurrentErrorHandler<\/a><\/noindex>, millega saab, mitte nihutades offsetit vea tekkimisel, s\u00f5numi hiljem t\u00f6\u00f6delda. Alates versioonist spring-kafka 2.3 on v\u00f5imalik m\u00e4\u00e4rata BackOffPolicy.<\/p>\n

<\/p>\n

See l\u00e4henemine v\u00f5imaldab uuesti t\u00f6\u00f6deldavatel s\u00f5numitel ellu j\u00e4\u00e4da rakenduse taask\u00e4ivitamise, 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\u00f5epoolest ei olnud seda mitu kuud rakenduse sellise uuesti t\u00f6\u00f6tlemise s\u00fcsteemiga). Ajutised vead p\u00f5hjustasid SeekToCurrentErrorHandler'i aktiveerimist. \u00dclej\u00e4\u00e4nud vead kirjutatakse logisse, p\u00f5hjustavad offseti nihkumist ja t\u00f6\u00f6tlemine j\u00e4tkub j\u00e4rgmise s\u00f5numiga.<\/p>\n

<\/p>\n

L\u00f5plik lahendus<\/h2>\n

<\/p>\n

SeekToCurrentErrorHandler'ile p\u00f5hinev rakendus viitis meid arendada oma mehhanismi s\u00f5numite edastamiseks uuesti.<\/p>\n

<\/p>\n

Esiteks soovisime kasutada juba olemasolevat teadmist ja laiendada seda rakenduse loogika p\u00f5hjal. Lineaarse loogikaga rakenduse jaoks oleks optimaalne peatada uute s\u00f5numite lugemine l\u00fchikeseks ajaks, mis on m\u00e4\u00e4ratud uuesti k\u00f5nede strateegia raames. \u00dclej\u00e4\u00e4nud rakenduste puhul soovisime omada \u00fchte punkti, mis tagaks uuesti k\u00f5nede strateegia elluviimise. Lisaks peaks see \u00fchtne punkt omama DLQ funktsionaalsust m\u00f5lema l\u00e4henemise jaoks.<\/p>\n

<\/p>\n

Uuesti k\u00f5nede strateegia peaks olema salvestatud rakenduses, mis vastutab j\u00e4rgmise intervalli saamise eest ajutiste vigade korral.<\/p>\n

<\/p>\n

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

<\/p>\n

Spring-kafka kasutamise korral v\u00f5ib Consumer'i peatamise kood v\u00e4lja n\u00e4ha 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-le\n    }<\/code><\/pre>\n
<\/p>\n
N\u00e4ites on retryAt aeg, millal tuleks uuesti k\u00e4ivitada MessageListenerContainer, kui see veel t\u00f6\u00f6tab. Uuesti k\u00e4ivitamine toimub eraldi l\u00f5imes, mis on k\u00e4ivitatud TaskScheduler'is, mille rakenduse pakub samuti spring. <\/p>\n
<\/p>\n
L\u00f6ime retryAt v\u00e4\u00e4rtuse j\u00e4rgmise meetodi abil:<\/p>\n
<\/p>\n
    \n
  1. Otsitakse uuesti k\u00f5nede arvu v\u00e4\u00e4rtust.<\/li>\n
  2. Vastavalt uuesti k\u00f5nede arvu v\u00e4\u00e4rtusele otsitakse praegune viivituse intervall uuesti k\u00f5nede strateegias. Strateegia kuulutatakse v\u00e4lja rakenduses, mille salvestamiseks valisime JSON formaadi.<\/li>\n
  3. JSON-massis leitud intervall sisaldab sekundeid, mille jooksul tuleb t\u00f6\u00f6tlemine uuesti korrata. See sekundite arv liidetakse praegusele ajale, moodustades retryAt v\u00e4\u00e4rtuse.<\/li>\n
  4. Kui intervalli ei leita, siis retryAt v\u00e4\u00e4rtus on null ja s\u00f5num saadetakse DLQ-sse k\u00e4sitsi t\u00f6\u00f6tlemiseks.<\/li>\n<\/ol>\n
    <\/p>\n
    Selle l\u00e4henemise puhul on vajalik s\u00e4ilitada igas s\u00f5numis, mis on praegu t\u00f6\u00f6tlemisel, korduvate katsete arv, n\u00e4iteks rakenduse m\u00e4lu. Katsete arvu s\u00e4ilitamine m\u00e4lus ei ole kriitiline, kuna lineaarse loogikaga rakendus ei pruugi kogu t\u00f6\u00f6tlemist korraldada. Erinevalt spring-retry'st ei too rakenduse taask\u00e4ivitamine kaasa k\u00f5igi s\u00f5numite kadumist uuesti t\u00f6\u00f6tlemiseks, vaid lihtsalt strateegia taask\u00e4ivitamise. <\/p>\n
    <\/p>\n
    See l\u00e4henemine aitab v\u00e4hendada koormust v\u00e4lisele s\u00fcsteemile, mis v\u00f5ib olla v\u00e4ga suure koormuse t\u00f5ttu k\u00e4ttesaamatu. Teisis\u00f5nu, lisaks uuesti t\u00f6\u00f6tlemisele oleme saavutanud mustri elluviimise. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    Meie puhul on veapiirang vaid 1, ja et minimeerida s\u00fcsteemi seiskumist ajutiste v\u00f5rgu katkestuste t\u00f5ttu, kasutame v\u00e4ga granulaarses korduste strateegias l\u00fchikesi viivitusintervalle. See v\u00f5ib mitte sobida k\u00f5igile ettev\u00f5tte grupi rakendustele, seet\u00f5ttu tuleb veapiirangu ja intervalli suuruse suhet valida, tuginedes s\u00fcsteemi omadustele.<\/p>\n
    <\/p>\n
    Eraldi rakendus s\u00f5numite t\u00f6\u00f6tlemiseks rakendustelt, millel on m\u00e4\u00e4ramatud loogikad.<\/h3>\n
    <\/p>\n
    Siin on n\u00e4ide koodist, mis saadab s\u00f5numi sellesse rakendusse (Retryer), mis t\u00e4idab uuesti saatmise DESTINATION teema juurde, kui RETRY_AT aeg on k\u00e4es:<\/p>\n
    <\/p>\n
    \npublic  void retry(ConsumerRecord record, String retryToTopic, \n                         Instant retryAt, String counter, String groupId, Exception e) {\n        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());\n        List
     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 messageToSend =\n            new ProducerRecord(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);\n        kafkaTemplate.send(messageToSend);\n    }<\/code><\/pre>\n
    <\/p>\n
    N\u00e4idatud n\u00e4itest on n\u00e4ha, et palju teavet edastatakse pealkirjades. RETRY_AT v\u00e4\u00e4rtus on samuti sama, mis Consumer\u2019i peatamise kordusmehhanismi puhul. Lisaks DESTINATIONile ja RETRY_AT-ile edastame:<\/p>\n
    <\/p>\n