{"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":"Rip\u00ebrpunimi i ngjarjeve t\u00eb marra nga Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Rip\u00ebrpunimi<\/p>\n

<\/p>\n

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

<\/p>\n

S\u00eb fundi kam ndar\u00eb p\u00ebrvoj\u00ebn<\/a><\/noindex> n\u00eb lidhje me parametrat q\u00eb ne si ekip shpesh p\u00ebrdorim p\u00ebr Kafka Producer dhe Consumer, p\u00ebr t'u afruar me dor\u00ebzimin e garantuar. N\u00eb k\u00ebt\u00eb artikull d\u00ebshiroj t\u00eb tregoj se si organizuam riprocesimin e nj\u00eb ngjarjeje t\u00eb marr\u00eb nga Kafka, si rezultat i pap\u00ebrshtatshm\u00ebris\u00eb s\u00eb nj\u00eb sistemi t\u00eb jasht\u00ebm.<\/p>\n

<\/p>\n

Aplikacionet moderne funksionojn\u00eb n\u00eb nj\u00eb mjedis shum\u00eb t\u00eb komplikuar. Logjika e biznesit, e mb\u00ebshtjell\u00eb n\u00eb nj\u00eb grumbull teknologjik modern, q\u00eb punon n\u00eb nj\u00eb imazh Docker, i menaxhuar nga nj\u00eb orkestrator si Kubernetes ose OpenShift, dhe q\u00eb komunikon me aplikacione t\u00eb tjera ose zgjidhje enterprise p\u00ebrmes nj\u00eb vargu ruterash fizik dhe virtual. N\u00eb nj\u00eb mjedis t\u00eb till\u00eb, gjithmon\u00eb mund t\u00eb ndodhin probleme, prandaj riprocesimi i ngjarjeve n\u00eb rastin e pap\u00ebrshtatshm\u00ebris\u00eb s\u00eb nj\u00eb 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\u00ebrdorim IBM MQ p\u00ebr dor\u00ebzimin e mesazheve n\u00eb m\u00ebnyr\u00eb asinkrone. Kur ndodhte ndonj\u00eb gabim gjat\u00eb pun\u00ebs s\u00eb sh\u00ebrbimit, mesazhi i marr\u00eb mund t\u00eb vendosej n\u00eb radh\u00ebn e mesazheve t\u00eb vdekura (DLQ) p\u00ebr shqyrtim t\u00eb m\u00ebtejsh\u00ebm. DLQ krijohej pran\u00eb radh\u00ebs hyr\u00ebse, dhe kalimi i mesazhit ndodhte brenda IBM MQ. <\/p>\n

<\/p>\n

N\u00ebse gabimi kishte nj\u00eb natyr\u00eb t\u00eb p\u00ebrkohshme dhe ne mund ta p\u00ebrcaktonim k\u00ebt\u00eb (p\u00ebr shembull, ResourceAccessException gjat\u00eb nj\u00eb thirrjeje HTTP ose MongoTimeoutException gjat\u00eb nj\u00eb k\u00ebrkese n\u00eb MongoDb), at\u00ebher\u00eb strategjia e rip\u00ebrs\u00ebritjeve hynte n\u00eb fuqi. Pavar\u00ebsisht nga ndarjet e logjik\u00ebs s\u00eb aplikacionit, mesazhi i origjin\u00ebs transferohej ose n\u00eb radh\u00ebn sistemike p\u00ebr d\u00ebrgim t\u00eb vonuar, ose n\u00eb nj\u00eb aplikacion t\u00eb ve\u00e7ant\u00eb, i cili u krijua vite m\u00eb par\u00eb p\u00ebr rip\u00ebrs\u00ebritjen e mesazheve. N\u00eb k\u00ebt\u00eb rast, n\u00eb kok\u00ebn e mesazhit shkruhej numri i rip\u00ebrs\u00ebritjes, i lidhur me intervalin e vones\u00ebs ose me fundin e strategjis\u00eb n\u00eb nivelin e aplikacionit. N\u00ebse arrijm\u00eb fundin e strategjis\u00eb, por sistemi i jasht\u00ebm ende nuk \u00ebsht\u00eb i disponuesh\u00ebm, at\u00ebher\u00eb mesazhi do t\u00eb vendoset n\u00eb DLQ p\u00ebr shqyrtim manual.<\/p>\n

<\/p>\n

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

<\/p>\n

Duke k\u00ebrkuar n\u00eb internet<\/a><\/noindex>, mund t\u00eb gjejm\u00eb k\u00ebt\u00eb vendim<\/a><\/noindex>. N\u00ebse e shkurtojm\u00eb, propozohet t\u00eb krijohet nj\u00eb tem\u00eb p\u00ebr \u00e7do interval vonese dhe t\u00eb implementohen n\u00eb an\u00ebn e aplikacionit Consumer q\u00eb do t\u00eb lexojn\u00eb mesazhet me vones\u00ebn e nevojshme. <\/p>\n

<\/p>\n

\"Rip\u00ebrpunimi<\/p>\n

<\/p>\n

Megjith\u00ebse ka nj\u00eb num\u00ebr t\u00eb madh komentesh pozitive, m\u00eb duket se nuk \u00ebsht\u00eb krejt\u00ebsisht e suksesshme. S\u00eb pari, sepse zhvilluesi, p\u00ebrve\u00e7 realizimit t\u00eb k\u00ebrkesave t\u00eb biznesit, do t\u00eb duhet t\u00eb shpenzoj\u00eb shum\u00eb koh\u00eb p\u00ebr t\u00eb realizuar mekanizmin e p\u00ebrshkruar.<\/p>\n

<\/p>\n

P\u00ebr m\u00eb tep\u00ebr, n\u00ebse menaxhimi i qasjes \u00ebsht\u00eb i aktivizuar n\u00eb klasterin Kafka, do t'i duhet t\u00eb investoj\u00eb nj\u00eb koh\u00eb p\u00ebr t\u00eb krijuar temat dhe p\u00ebr t\u00eb siguruar qasjet e nevojshme p\u00ebr to. N\u00eb p\u00ebrve\u00e7im t\u00eb k\u00ebsaj, do t\u00eb duhet t\u00eb zgjidhni parametrin e duhur retention.ms p\u00ebr \u00e7do nga temat e riprovimit, n\u00eb m\u00ebnyr\u00eb q\u00eb mesazhet t\u00eb d\u00ebrgohen p\u00ebrs\u00ebri dhe t\u00eb mos humbasin. Zbatimi dhe k\u00ebrkesa e qasjeve 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 shohim se cilat mekanizma p\u00ebr p\u00ebrpunimin e p\u00ebrs\u00ebritur t\u00eb mesazheve na ofron spring-i n\u00eb p\u00ebrgjith\u00ebsi dhe spring-kafka n\u00eb ve\u00e7anti. Spring-kafka ka nj\u00eb var\u00ebsi tranzitive mbi spring-retry, i cili ofron abstraksione p\u00ebr menaxhimin e politikat e BackOff. Kjo \u00ebsht\u00eb nj\u00eb mjet mjaft fleksib\u00ebl, por nj\u00eb nga disavantazhet e tij t\u00eb r\u00ebnd\u00ebsishme \u00ebsht\u00eb ruajtja e mesazheve p\u00ebr rip\u00ebrpunim n\u00eb memorjen e aplikacionit. Kjo do t\u00eb thot\u00eb se rinisja e aplikacionit p\u00ebr shkak t\u00eb azhurnimit ose gabimeve gjat\u00eb p\u00ebrdorimit do t\u00eb \u00e7oj\u00eb n\u00eb humbjen e t\u00eb gjith\u00eb mesazheve q\u00eb presin p\u00ebr rip\u00ebrpunim. Duke qen\u00eb se ky pik\u00eb \u00ebsht\u00eb kritik p\u00ebr sistemin ton\u00eb, ne nuk e konsideruam m\u00eb tej.<\/p>\n

<\/p>\n

Vet\u00eb spring-kafka ofron disa zbatime t\u00eb ContainerAwareErrorHandler, p\u00ebr shembull, SeekToCurrentErrorHandler<\/a><\/noindex>, me ndihm\u00ebn e s\u00eb cil\u00ebs mund t\u00eb p\u00ebrpunoni nj\u00eb mesazh m\u00eb von\u00eb pa nd\u00ebrruar offset n\u00eb rast gabimi. Prej versionit spring-kafka 2.3, \u00ebsht\u00eb b\u00ebr\u00eb e mundur t\u00eb p\u00ebrcaktojn\u00eb BackOffPolicy.<\/p>\n

<\/p>\n

Ky qasje lejon q\u00eb mesazhet q\u00eb rip\u00ebrpunohen t\u00eb p\u00ebrjetojn\u00eb rinisjen e aplikacionit, por mekanizmi DLQ ende mungon. Ky variant ishte ai q\u00eb ne zgjodh\u00ebm n\u00eb fillim t\u00eb vitit 2019, duke qen\u00eb optimist se DLQ nuk do t\u00eb ishte e nevojshme (na ndihmoi dhe me t\u00eb v\u00ebrtet\u00eb nuk u nevojit p\u00ebr disa muaj shfryt\u00ebzimi t\u00eb aplikacionit me nj\u00eb sistem t\u00eb till\u00eb t\u00eb rip\u00ebrpunimit). Gabimet temporale \u00e7onin n\u00eb aktivizimin e SeekToCurrentErrorHandler. Gabimet e tjera shfaqeshin n\u00eb log, \u00e7onin n\u00eb nd\u00ebrrimin e offset, dhe p\u00ebrpunimi vazhdonte me mesazhin tjet\u00ebr.<\/p>\n

<\/p>\n

Zgjidhja p\u00ebrfundimtare<\/h2>\n

<\/p>\n

Implementimi i bazuar n\u00eb SeekToCurrentErrorHandler na shtyu t\u00eb zhvillonim nj\u00eb mekaniz\u00ebm t\u00eb vetin p\u00ebr ri-d\u00ebrgimin e mesazheve.<\/p>\n

<\/p>\n

S\u00eb pari, ne doja t\u00eb shfryt\u00ebzojm\u00eb p\u00ebrvoj\u00ebn ekzistuese dhe ta zgjeronim at\u00eb n\u00eb var\u00ebsi t\u00eb logjik\u00ebs s\u00eb aplikacionit. P\u00ebr nj\u00eb aplikacion me logjik\u00eb lineare, optimale do t\u00eb ishte t\u00eb ndalosh leximin e mesazheve t\u00eb reja p\u00ebr nj\u00eb periudh\u00eb t\u00eb vog\u00ebl kohore, t\u00eb caktuar n\u00eb kuad\u00ebr t\u00eb strategjis\u00eb s\u00eb ri-thirrjeve. P\u00ebr aplikacionet e tjera, do t\u00eb d\u00ebshironim t\u00eb kishim nj\u00eb pik\u00eb t\u00eb vetme q\u00eb do t\u00eb garantonte p\u00ebrmbushjen e strategjis\u00eb s\u00eb ri-thirrjeve. P\u00ebrve\u00e7 k\u00ebsaj, kjo pik\u00eb e vetme duhet t\u00eb ket\u00eb funksionalitetin e DLQ p\u00ebr t\u00eb dyja qasjet.<\/p>\n

<\/p>\n

Strategjia e ri-thirrjeve duhet t\u00eb ruhet n\u00eb aplikacionin q\u00eb p\u00ebrgjigjet p\u00ebr t\u00eb marr\u00eb intervalin e ardhsh\u00ebm n\u00eb rast t\u00eb nj\u00eb gabimi t\u00eb p\u00ebrkohsh\u00ebm.<\/p>\n

<\/p>\n

Ndalesa e Consumer-it p\u00ebr nj\u00eb aplikacion me logjik\u00eb lineare<\/h3>\n

<\/p>\n

Kur punoni me spring-kafka, kodi p\u00ebr ndalimin e Consumer-it mund t\u00eb duket di\u00e7ka si kjo:<\/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        \/\/ n\u00eb DLQ\n    }<\/code><\/pre>\n
<\/p>\n
N\u00eb shembullin e m\u00ebsip\u00ebrm, retryAt \u00ebsht\u00eb koha kur duhet t\u00eb rifilloni MessageListenerContainer-n, n\u00ebse ai \u00ebsht\u00eb ende duke punuar. Rihapja do t\u00eb ndodh\u00eb n\u00eb nj\u00eb thread t\u00eb ve\u00e7ant\u00eb, i nisur n\u00eb TaskScheduler, implementimin e t\u00eb cilit gjithashtu e ofron spring. <\/p>\n
<\/p>\n
Vlera retryAt ne e gjejm\u00eb n\u00eb k\u00ebt\u00eb m\u00ebnyr\u00eb:<\/p>\n
<\/p>\n
    \n
  1. K\u00ebrkohet vlera e num\u00ebruesit t\u00eb ri-thirrjeve.<\/li>\n
  2. N\u00eb p\u00ebrputhje me vler\u00ebn e num\u00ebruesit, k\u00ebrkohet intervali aktual i vones\u00ebs n\u00eb strategjin\u00eb e ri-thirrjeve. Strategjia shpallet brenda aplikacionit, 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 duhet t\u00eb p\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, at\u00ebher\u00eb 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\u00ebtij qasje, mbetet vet\u00ebm t\u00eb ruhet numri i p\u00ebrpjekjeve p\u00ebr \u00e7do mesazh q\u00eb \u00ebsht\u00eb tani n\u00eb procesim, p\u00ebr shembull n\u00eb memorien e aplikacionit. Ruajtja e numrit t\u00eb p\u00ebrpjekjeve n\u00eb memorien e aplikacionit nuk \u00ebsht\u00eb kritike p\u00ebr k\u00ebt\u00eb qasje, pasi aplikacioni me logjik\u00eb lineare nuk mund t\u00eb p\u00ebrpunoj\u00eb gjith\u00e7ka. N\u00eb krahasim me spring-retry, riparagjykimi i aplikacionit nuk do t\u00eb \u00e7oj\u00eb n\u00eb humbjen e t\u00eb gjitha mesazheve p\u00ebr rip\u00ebrpunim, por thjesht n\u00eb rinisjen e strategjis\u00eb. <\/p>\n
    <\/p>\n
    Kjo qasje ndihmon n\u00eb leht\u00ebsimin e ngarkes\u00ebs nga sistemi i jasht\u00ebm, i cili mund t\u00eb jet\u00eb i paaksesuesh\u00ebm p\u00ebr shkak t\u00eb ngarkes\u00ebs shum\u00eb t\u00eb madhe. N\u00eb fjal\u00eb t\u00eb tjera, p\u00ebrve\u00e7 rip\u00ebrpunimit, kemi arritur n\u00eb 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 koh\u00ebn e vones\u00ebs s\u00eb sistemit p\u00ebr shkak t\u00eb nj\u00eb nd\u00ebrprerjeje t\u00eb p\u00ebrkohshme rrjetore, ne p\u00ebrdorim nj\u00eb strategji shum\u00eb granulare t\u00eb p\u00ebrpjekjeve t\u00eb rip\u00ebrs\u00ebritura me intervale t\u00eb vogla vonese. Kjo mund t\u00eb mos jet\u00eb e p\u00ebrshtatshme p\u00ebr t\u00eb gjitha aplikacionet e grupit, prandaj raporti midis pragut t\u00eb gabimit dhe madh\u00ebsis\u00eb s\u00eb intervalit duhet t\u00eb p\u00ebrcaktohet duke u bazuar n\u00eb karakteristikat e sistemit.<\/p>\n
    <\/p>\n
    Nj\u00eb aplikacion t\u00eb ve\u00e7ant\u00eb p\u00ebr t\u00eb p\u00ebrpunuar mesazhet nga aplikacionet me logjik\u00eb t\u00eb pabesueshme<\/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 kryej\u00eb rip\u00ebrs\u00ebritjen n\u00eb tem\u00ebn DESTINATION kur t\u00eb arrij\u00eb koha RETRY_AT:<\/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
    Nga shembulli duket se shum\u00eb informacion p\u00ebrcillet n\u00eb header. Vlera RETRY_AT \u00ebsht\u00eb e nj\u00ebjt\u00eb si p\u00ebr mekanizmin e p\u00ebrs\u00ebritjes p\u00ebrmes ndalimit t\u00eb Consumer\u2019it. P\u00ebrve\u00e7 DESTINATION dhe RETRY_AT, ne kalojm\u00eb:<\/p>\n
    <\/p>\n