{"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\/sq\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Riprocesimi i ngjarjeve t\u00eb marra nga Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Riprocesimi<\/p>\n

<\/p>\n

P\u00ebrsh\u00ebndetje, Habr.<\/p>\n

<\/p>\n

S\u00eb fundmi un\u00eb ndava p\u00ebrvoj\u00eb<\/a><\/noindex> rreth parametrave q\u00eb ekipi yn\u00eb shpesh i p\u00ebrdor p\u00ebr Kafka Producer dhe Consumer, p\u00ebr t'u afruar pran\u00eb d\u00ebrges\u00ebs s\u00eb garantuar. N\u00eb k\u00ebt\u00eb artikull d\u00ebshiroj t\u00eb flas p\u00ebr m\u00ebnyr\u00ebn se si organizuam riciklimin e ngjarjeve, t\u00eb marra nga Kafka, si rezultat i papjeshm\u00ebris\u00eb p\u00ebr koh\u00eb nga nj\u00eb sistem t\u00eb jasht\u00ebm.<\/p>\n

<\/p>\n

Aplikacionet moderne funksionojn\u00eb n\u00eb nj\u00eb ambient shum\u00eb kompleks. Logjika e biznesit, e mb\u00ebshtjell\u00eb n\u00eb nj\u00eb tech stack modern, duke funksionuar n\u00eb nj\u00eb imazh Docker, i menaxhuar nga nj\u00eb orkestrator si Kubernetes ose OpenShift, dhe duke komunikuar me aplikacione t\u00eb tjera ose zgjidhje enterprise p\u00ebrmes nj\u00eb zinxhiri ruterash fizik\u00eb dhe virtual\u00eb. N\u00eb nj\u00eb ambient t\u00eb till\u00eb gjithmon\u00eb di\u00e7ka mund t\u00eb prishet, prandaj riciklimi i ngjarjeve n\u00eb rastin e papjeshm\u00ebris\u00eb s\u00eb nj\u00ebrit prej sistemeve t\u00eb jashtme \u00ebsht\u00eb nj\u00eb pjes\u00eb e r\u00ebnd\u00ebsishme e proceseve tona t\u00eb biznesit.<\/p>\n

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

Si ishte para Kafka<\/h2>\n

<\/p>\n

M\u00eb par\u00eb n\u00eb projekt ne p\u00ebrdornim IBM MQ p\u00ebr d\u00ebrgimin asinkron t\u00eb mesazheve. Kur ndodhte ndonj\u00eb gabim n\u00eb procesin e funksionimit t\u00eb sh\u00ebrbimit, mesazhi i marr\u00eb mund t\u00eb vendosej n\u00eb dead-letter-queue (DLQ) p\u00ebr shqyrtim manual. DLQ krijohej pran\u00eb radh\u00ebs hyr\u00ebse, dhe zhvendosja e mesazheve ndodhte brenda IBM MQ. <\/p>\n

<\/p>\n

N\u00ebse gabimi kishte natyr\u00eb t\u00eb p\u00ebrkohshme dhe ne mund ta p\u00ebrcaktonim k\u00ebt\u00eb (p.sh., ResourceAccessException gjat\u00eb thirrjes HTTP ose MongoTimeoutException gjat\u00eb k\u00ebrkes\u00ebs n\u00eb MongoDb), at\u00ebher\u00eb hyhej n\u00eb strategjin\u00eb e rip\u00ebrpjekjeve. Pavar\u00ebsisht nga deg\u00ebzimi i logjik\u00ebs s\u00eb aplikacionit, mesazhi origjinal zhvendosej ose n\u00eb radh\u00ebn sistematike p\u00ebr d\u00ebrges\u00eb t\u00eb vonuar, ose n\u00eb nj\u00eb aplikacion t\u00eb ve\u00e7ant\u00eb q\u00eb ishte b\u00ebr\u00eb dikur p\u00ebr rip\u00ebrs\u00ebritjen e mesazheve. Gjat\u00eb k\u00ebsaj periudhe, n\u00eb titullin e mesazhit ruheshin numri i rip\u00ebrs\u00ebritjeve, i lidhur me intervalin e vones\u00ebs ose me fundin e strategjis\u00eb n\u00eb nivel aplikacioni. N\u00ebse arrijm\u00eb n\u00eb fundin e strategjis\u00eb, por sistemi i jasht\u00ebm ende nuk \u00ebsht\u00eb i aksesuesh\u00ebm, mesazhi do t\u00eb vendoset n\u00eb DLQ p\u00ebr shqyrtim manual.<\/p>\n

<\/p>\n

K\u00ebrkimi i zgjidhjes<\/h2>\n

<\/p>\n

Pas k\u00ebrkimit n\u00eb internet<\/a><\/noindex>, mund t\u00eb gjenden k\u00ebto zgjidhje<\/a><\/noindex>. N\u00ebse e p\u00ebrmbledhim, propozohet t\u00eb krijohet nj\u00eb tem\u00eb p\u00ebr \u00e7do interval vonese dhe t\u00eb zbatohen n\u00eb an\u00ebn e aplikacionit konsumator\u00ebt q\u00eb do t\u00eb lexojn\u00eb mesazhet me vones\u00ebn e nevojshme. <\/p>\n

<\/p>\n

\"Riprocesimi<\/p>\n

<\/p>\n

Pavar\u00ebsisht numrit t\u00eb madh t\u00eb komenteve pozitive, duket se nuk \u00ebsht\u00eb shum\u00eb e suksesshme. N\u00eb radh\u00eb t\u00eb par\u00eb p\u00ebr shkak se zhvilluesi, p\u00ebrve\u00e7 zbatimit t\u00eb k\u00ebrkesave t\u00eb biznesit, do t\u00eb duhet t\u00eb kaloj\u00eb shum\u00eb koh\u00eb n\u00eb zbatimin e mekanizmit t\u00eb p\u00ebrshkruar.<\/p>\n

<\/p>\n

P\u00ebr m\u00eb tep\u00ebr, n\u00ebse menaxhimi i aksesit \u00ebsht\u00eb i aktivizuar n\u00eb klasterin Kafka, do t\u00eb duhet t\u00eb kaloni nj\u00eb koh\u00eb p\u00ebr t\u00eb krijuar temat dhe p\u00ebr t\u00eb siguruar akseset e nevojshme p\u00ebr to. P\u00ebrve\u00e7 k\u00ebsaj, do t\u00eb jet\u00eb e nevojshme t\u00eb p\u00ebrcaktohet parametrin e duhur retention.ms p\u00ebr \u00e7do tem\u00eb t\u00eb riangazhimit, n\u00eb m\u00ebnyr\u00eb q\u00eb mesazhet t\u00eb d\u00ebrgohen p\u00ebrs\u00ebri dhe t\u00eb mos humbasin prej saj. Zbatimi dhe k\u00ebrkesa p\u00ebr akses do t\u00eb duhet t\u00eb p\u00ebrs\u00ebritet p\u00ebr \u00e7do sh\u00ebrbim ekzistues ose t\u00eb ri.<\/p>\n

<\/p>\n

Tani le t\u00eb shikojm\u00eb se cilat mekanizma p\u00ebr p\u00ebrpunimin e p\u00ebrs\u00ebritur t\u00eb mesazheve na ofron spring n\u00eb p\u00ebrgjith\u00ebsi dhe spring-kafka n\u00eb ve\u00e7anti. Spring-kafka ka nj\u00eb var\u00ebsi transitive n\u00eb spring-retry, e cila ofron abstrahqione p\u00ebr menaxhimin e politikave t\u00eb ndryshme t\u00eb BackOff. Ky \u00ebsht\u00eb nj\u00eb mjet mjaft fleksib\u00ebl, por nj\u00eb nga disavantazhet e saj t\u00eb r\u00ebnd\u00ebsishme \u00ebsht\u00eb ruajtja e mesazheve p\u00ebr riciklim n\u00eb memorien e aplikacionit. Kjo do t\u00eb thot\u00eb se rinisja e aplikacionit p\u00ebr shkak t\u00eb nj\u00eb p\u00ebrdit\u00ebsimi ose gabimi gjat\u00eb operimit do t\u00eb sjell\u00eb humbjen e t\u00eb gjitha mesazheve q\u00eb presin p\u00ebrpunim t\u00eb p\u00ebrs\u00ebritur. Pasi ky pik\u00eb \u00ebsht\u00eb kritik p\u00ebr sistemin ton\u00eb, ne nuk vendos\u00ebm ta shqyrtojm\u00eb m\u00eb tej.<\/p>\n

<\/p>\n

Vet\u00eb spring-kafka ofron disa realizime t\u00eb ContainerAwareErrorHandler, p\u00ebr shembull SeekToCurrentErrorHandler<\/a><\/noindex>, me an\u00eb t\u00eb t\u00eb cilit mund t\u00eb p\u00ebrpunoni mesazhin m\u00eb von\u00eb, pa e l\u00ebvizur offset-in n\u00eb rast t\u00eb nj\u00eb gabimi. Duke filluar nga versioni spring-kafka 2.3, u b\u00eb e mundur t\u00eb p\u00ebrcaktoni BackOffPolicy.<\/p>\n

<\/p>\n

Ky qasje lejon q\u00eb mesazhet e riprocesueshme t\u00eb p\u00ebrjetojn\u00eb rinisjen e aplikacionit, por mekanizmi DLQ ende mungon. Ky \u00ebsht\u00eb opsioni q\u00eb ne zgjodh\u00ebm n\u00eb fillim t\u00eb vitit 2019, me optimizmin se DLQ nuk do t\u00eb ishte e nevojshme (na ka p\u00ebrkufizuar dhe me t\u00eb v\u00ebrtet\u00eb nuk u desh p\u00ebr disa muaj shfryt\u00ebzimi t\u00eb aplikacionit me k\u00ebt\u00eb sistem riprocesimi). Gabimet p\u00ebrkoh\u00ebsore e shkaktuan aktivizimin e SeekToCurrentErrorHandler. Gabimet e tjera u regjistruan n\u00eb log, shkaktuan zhvendosjen e offset, dhe p\u00ebrpunimi vazhdoi me mesazhin tjet\u00ebr.<\/p>\n

<\/p>\n

Zgjidhja p\u00ebrfundimtare<\/h2>\n

<\/p>\n

Realizimi, i bazuar n\u00eb SeekToCurrentErrorHandler, na nxit\u00ebn t\u00eb zhvillojm\u00eb mekanizmin ton\u00eb p\u00ebr ricel\u00ebsimin e mesazheve.<\/p>\n

<\/p>\n

N\u00eb radh\u00eb t\u00eb par\u00eb, ne d\u00ebshironim t\u00eb shfryt\u00ebzonim p\u00ebrvoj\u00ebn ekzistuese dhe ta zgjeronim at\u00eb sipas logjik\u00ebs s\u00eb aplikacionit. P\u00ebr nj\u00eb aplikacion me logjik\u00eb lineare, do t\u00eb ishte optimale t\u00eb ndalosh leximin e mesazheve t\u00eb reja p\u00ebr nj\u00eb periudh\u00eb t\u00eb vog\u00ebl kohe, t\u00eb caktuar n\u00eb kuad\u00ebr t\u00eb strategjis\u00eb s\u00eb rip\u00ebrs\u00ebritjes. P\u00ebr aplikacionet e tjera, do t\u00eb d\u00ebshironim nj\u00eb pik\u00eb t\u00eb vetme q\u00eb do t\u00eb siguroj\u00eb zbatimin e strategjis\u00eb s\u00eb rip\u00ebrs\u00ebritjes. P\u00ebr m\u00eb tep\u00ebr, kjo pik\u00eb e vetme duhet t\u00eb ket\u00eb funksionalitetin DLQ p\u00ebr t\u00eb dy qasjet.<\/p>\n

<\/p>\n

Strategjia e vet\u00eb rip\u00ebrs\u00ebritjes duhet t\u00eb ruhet n\u00eb aplikacionin q\u00eb \u00ebsht\u00eb p\u00ebrgjegj\u00ebs p\u00ebr marrjen e intervalit t\u00eb ardhsh\u00ebm n\u00eb rast t\u00eb nj\u00eb gabimi temporal.<\/p>\n

<\/p>\n

Ndalimi i Consumer-it p\u00ebr aplikacionin me logjik\u00eb lineare<\/h3>\n

<\/p>\n

Kur punoni me spring-kafka, kodi p\u00ebr ndalimin e Consumer-it mund t\u00eb duket paksa k\u00ebshtu:<\/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        \/\/ te DLQ\n    }<\/code><\/pre>\n
<\/p>\n
N\u00eb shembullin retryAt, \u00ebsht\u00eb koha kur duhet t\u00eb rifilloni p\u00ebrs\u00ebri MessageListenerContainer n\u00ebse ai \u00ebsht\u00eb akoma n\u00eb funksion. Rifillimi do t\u00eb ndodh\u00eb n\u00eb nj\u00eb thread t\u00eb ve\u00e7ant\u00eb, i cili do t\u00eb nis\u00eb n\u00eb TaskScheduler, zbatimin e t\u00eb cilit e ofron gjithashtu spring. <\/p>\n
<\/p>\n
Vlera retryAt gjendet n\u00eb k\u00ebt\u00eb m\u00ebnyr\u00eb:<\/p>\n
<\/p>\n
    \n
  1. K\u00ebrkohet vlera e num\u00ebruesit t\u00eb rip\u00ebrs\u00ebritjeve.<\/li>\n
  2. Pavar\u00ebsisht nga vlera e num\u00ebruesit, k\u00ebrkohet intervali aktual i vones\u00ebs n\u00eb strategjin\u00eb e rip\u00ebrs\u00ebritjeve. Strategjia shpallet n\u00eb aplikacionin vet\u00eb, dhe p\u00ebr ruajtjen e saj ne zgjodh\u00ebm formatin JSON.<\/li>\n
  3. Intervali i gjetur n\u00eb masivin JSON p\u00ebrmban numrin e sekondave pas t\u00eb cilave do t\u00eb nevojitet t\u00eb rip\u00ebrs\u00ebritet p\u00ebrpunimi. Ky num\u00ebr sekondash shtohet n\u00eb koh\u00ebn aktuale, duke formuar vler\u00ebn p\u00ebr retryAt.<\/li>\n
  4. N\u00ebse intervali nuk gjendet, vlera retryAt \u00ebsht\u00eb null dhe mesazhi do t\u00eb d\u00ebrgohet n\u00eb DLQ p\u00ebr shqyrtim manual.<\/li>\n<\/ol>\n
    <\/p>\n
    Me k\u00ebt\u00eb qasje, mbetet vet\u00ebm t\u00eb ruhet numri i thirrjeve t\u00eb p\u00ebrs\u00ebritura p\u00ebr \u00e7do mesazh q\u00eb \u00ebsht\u00eb aktualisht n\u00eb proces, p\u00ebr shembull n\u00eb memorien e aplikacionit. Ruajtja e numrit t\u00eb p\u00ebrpjekjeve n\u00eb memorie nuk \u00ebsht\u00eb kritike p\u00ebr k\u00ebt\u00eb qasje, pasi aplikacioni me logjik\u00eb lineare nuk mund t\u00eb kryej\u00eb p\u00ebrpunimin n\u00eb t\u00ebr\u00ebsi. Ndryshe nga spring-retry, rind\u00ebrtimi i aplikacionit nuk do t\u00eb \u00e7oj\u00eb n\u00eb humbjen e t\u00eb gjith\u00eb mesazheve p\u00ebr p\u00ebrs\u00ebritje, por thjesht do t\u00eb rinis\u00eb strategjin\u00eb. <\/p>\n
    <\/p>\n
    Kjo qasje ndihmon n\u00eb leht\u00ebsimin e ngarkes\u00ebs n\u00eb sistemin e jasht\u00ebm, i cili mund t\u00eb jet\u00eb i paaksesuesh\u00ebm p\u00ebr shkak t\u00eb nj\u00eb ngarkese shum\u00eb t\u00eb lart\u00eb. N\u00eb fjal\u00eb t\u00eb tjera, p\u00ebrve\u00e7 p\u00ebrpunimit t\u00eb p\u00ebrs\u00ebritur, ne arrit\u00ebm implementimin e modelit. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    N\u00eb rastin ton\u00eb, pragu i gabimit \u00ebsht\u00eb vet\u00ebm 1, dhe p\u00ebr t\u00eb minimizuar pezullimin e sistemit p\u00ebr shkak t\u00eb nj\u00eb nd\u00ebrprerjeje t\u00eb p\u00ebrkohshme rrjetore, ne p\u00ebrdorim nj\u00eb strategji shum\u00eb t\u00eb detajuar p\u00ebr thirrjet e p\u00ebrs\u00ebritura me intervale t\u00eb vogla vonesash. Kjo mund t\u00eb mos jet\u00eb e p\u00ebrshtatshme p\u00ebr t\u00eb gjith\u00eb aplikacionet e grupit t\u00eb kompanive, prandaj raporti midis pragut t\u00eb gabimit dhe madh\u00ebsis\u00eb s\u00eb intervalit duhet t\u00eb p\u00ebrshtatet, duke u bazuar n\u00eb karakteristikat e sistemit.<\/p>\n
    <\/p>\n
    Nj\u00eb aplikacion i ve\u00e7ant\u00eb p\u00ebr p\u00ebrpunimin e mesazheve nga aplikacionet me logjik\u00eb t\u00eb pacaktuar.<\/h3>\n
    <\/p>\n
    Ja nj\u00eb shembull kodi q\u00eb d\u00ebrgon nj\u00eb mesazh n\u00eb nj\u00eb aplikacion t\u00eb till\u00eb (Retryer), i cili do t\u00eb p\u00ebrs\u00ebris\u00eb d\u00ebrgimin n\u00eb tem\u00ebn DESTINATION kur t\u00eb arrihet koha 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
    Nga shembulli shihet se shum\u00eb informacione d\u00ebrgohen n\u00eb headra. Vlera RETRY_AT \u00ebsht\u00eb e nj\u00ebjt\u00eb si p\u00ebr mekanizmin e p\u00ebrs\u00ebritjes p\u00ebrmes ndalimit t\u00eb Consumer-it. P\u00ebrve\u00e7 DESTINATION dhe RETRY_AT, ne d\u00ebrgojm\u00eb:<\/p>\n
    <\/p>\n