{"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\/hu\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Esem\u00e9nyek \u00fajrafeldolgoz\u00e1sa, amelyek a Kafk\u00e1t\u00f3l sz\u00e1rmaznak.","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Esem\u00e9nyek<\/p>\n

<\/p>\n

Hell\u00f3, Habr.<\/p>\n

<\/p>\n

Nemr\u00e9g megosztottam a tapasztalataimat arr\u00f3l, hogy milyen param\u00e9tereket haszn\u00e1lunk a csapatban a Kafka Producer \u00e9s Consumer eset\u00e9ben, hogy k\u00f6zel ker\u00fclj\u00fcnk a garant\u00e1lt k\u00e9zbes\u00edt\u00e9shez. Ebben a cikkben szeretn\u00e9m elmondani, hogyan szervezt\u00fck meg az esem\u00e9nyek \u00fajrafeldolgoz\u00e1s\u00e1t, amely a Kafka-b\u00f3l \u00e9rkezett egy k\u00fcls\u0151 rendszer ideiglenes el\u00e9rhetetlens\u00e9ge miatt t\u00f6rt\u00e9nt.<\/a><\/noindex> A modern alkalmaz\u00e1sok nagyon bonyolult k\u00f6rnyezetben m\u0171k\u00f6dnek. Az \u00fczleti logika, amely a modern technol\u00f3giai stack-be van csomagolva, Docker-k\u00e9pben fut, amelyet olyan orchestr\u00e1tor ir\u00e1ny\u00edt, mint a Kubernetes vagy az OpenShift, \u00e9s m\u00e1s alkalmaz\u00e1sokkal vagy v\u00e1llalati megold\u00e1sokkal kommunik\u00e1l a fizikai \u00e9s virtu\u00e1lis routerek l\u00e1ncolat\u00e1n kereszt\u00fcl. Ilyen k\u00f6rnyezetben mindig t\u00f6rt\u00e9nhet valami hiba, ez\u00e9rt az esem\u00e9nyek \u00fajrafeldolgoz\u00e1sa a k\u00fcls\u0151 rendszerek egyik\u00e9nek el\u00e9rhetetlens\u00e9ge eset\u00e9n fontos r\u00e9sze \u00fczleti folyamataik.<\/p>\n

<\/p>\n

A Kafka el\u0151tt<\/p>\n

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

Kor\u00e1bban a projektben IBM MQ-t haszn\u00e1ltunk az aszinkron \u00fczenetk\u00fcld\u00e9shez. Ha hiba t\u00f6rt\u00e9nt a szolg\u00e1ltat\u00e1s m\u0171k\u00f6d\u00e9se sor\u00e1n, a kapott \u00fczenet helyet cser\u00e9lhetett a dead-letter-queue (DLQ) sz\u00e1m\u00e1ra tov\u00e1bbi k\u00e9zi elemz\u00e9s c\u00e9lj\u00e1b\u00f3l. A DLQ az \u00e9rkez\u0151 queue mellett j\u00f6tt l\u00e9tre, az \u00fczenet \u00e1tkonvert\u00e1l\u00e1sa az IBM MQ-n bel\u00fcl t\u00f6rt\u00e9nt.<\/h2>\n

<\/p>\n

Ha a hiba \u00e1tmeneti volt \u00e9s ezt meg tudtuk hat\u00e1rozni (p\u00e9ld\u00e1ul ResourceAccessException HTTP-h\u00edv\u00e1skor vagy MongoTimeoutException MongoDb lek\u00e9rdez\u00e9sn\u00e9l), akkor l\u00e9pett \u00e9letbe az \u00fajrah\u00edv\u00e1si strat\u00e9gia. F\u00fcggetlen\u00fcl az alkalmaz\u00e1s logik\u00e1j\u00e1nak el\u00e1gaz\u00e1s\u00e1t\u00f3l, az eredeti \u00fczenet vagy a rendszerqueue-ba ker\u00fclt az elhalasztott k\u00fcld\u00e9shez, vagy egy k\u00fcl\u00f6n alkalmaz\u00e1sba, amelyet r\u00e9gen k\u00e9sz\u00edtettek \u00fczenetek \u00fajrak\u00fcld\u00e9s\u00e9re. Ek\u00f6zben az \u00fczenet fejl\u00e9c\u00e9be fel lett t\u00fcntetve az \u00fajrak\u00fcld\u00e9sek sz\u00e1ma, amely a k\u00e9sleltet\u00e9si intervallumhoz vagy az alkalmaz\u00e1s szintj\u00e9n l\u00e9v\u0151 strat\u00e9gia v\u00e9g\u00e9hez volt k\u00f6tve. Ha el\u00e9rt\u00fck a strat\u00e9gia v\u00e9g\u00e9t, de a k\u00fcls\u0151 rendszer m\u00e9g mindig nem el\u00e9rhet\u0151, az \u00fczenet a DLQ-be ker\u00fcl a k\u00e9zi elemz\u00e9shez. <\/p>\n

<\/p>\n

Megold\u00e1s keres\u00e9se<\/p>\n

<\/p>\n

Interneten keresg\u00e9lve<\/h2>\n

<\/p>\n

, a k\u00f6vetkez\u0151 tal\u00e1lhat\u00f3<\/a><\/noindex>. R\u00f6viden, azt javasolj\u00e1k, hogy minden k\u00e9sleltet\u00e9shez egy topikot hozzanak l\u00e9tre, \u00e9s az alkalmaz\u00e1s oldal\u00e1n Consumer-eket implement\u00e1ljanak, amelyek a sz\u00fcks\u00e9ges k\u00e9s\u00e9ssel olvass\u00e1k el az \u00fczeneteket. d\u00f6nt\u00e9s<\/a><\/noindex>. <\/p>\n

<\/p>\n

\"Esem\u00e9nyek<\/p>\n

<\/p>\n

B\u00e1r sok pozit\u00edv v\u00e9lem\u00e9nyt kaptam, sz\u00e1momra nem t\u0171nik teljesen sikeresnek. Els\u0151sorban az\u00e9rt, mert a fejleszt\u0151nek, a biznisz k\u00f6vetelm\u00e9nyek megval\u00f3s\u00edt\u00e1sa mellett, sok id\u0151t kell majd ford\u00edtania a le\u00edrt mechanizmus megval\u00f3s\u00edt\u00e1s\u00e1ra.<\/p>\n

<\/p>\n

Tov\u00e1bb\u00e1, ha a Kafka klaszteren enged\u00e9lyezve van a hozz\u00e1f\u00e9r\u00e9skezel\u00e9s, akkor valamennyi id\u0151t el kell t\u00f6lteni a t\u00e9m\u00e1k l\u00e9trehoz\u00e1s\u00e1val \u00e9s a sz\u00fcks\u00e9ges hozz\u00e1f\u00e9r\u00e9sek biztos\u00edt\u00e1s\u00e1val. Ezen k\u00edv\u00fcl meg kell tal\u00e1lni a megfelel\u0151 retention.ms param\u00e9tert minden egyes retry-t\u00e9m\u00e1hoz, hogy az \u00fczenetek id\u0151ben \u00fajra elk\u00fcld\u00e9sre ker\u00fclhessenek, \u00e9s ne t\u0171njenek el. A hozz\u00e1f\u00e9r\u00e9sek megval\u00f3s\u00edt\u00e1s\u00e1t \u00e9s ig\u00e9nyl\u00e9s\u00e9t minden megl\u00e9v\u0151 vagy \u00faj szolg\u00e1ltat\u00e1s eset\u00e9n meg kell ism\u00e9telni.<\/p>\n

<\/p>\n

N\u00e9zz\u00fck most meg, milyen mechanizmusokat k\u00edn\u00e1l a spring a visszafejleszt\u00e9shez \u00e1ltal\u00e1ban, \u00e9s a spring-kafka k\u00fcl\u00f6n\u00f6sen. A spring-kafka tranzit\u00edv f\u00fcgg\u0151s\u00e9get hoz l\u00e9tre a spring-retry-hoz, amely absztrakci\u00f3kat biztos\u00edt a k\u00fcl\u00f6nb\u00f6z\u0151 BackOffPolicy kezel\u00e9sekhez. Ez egy el\u00e9g rugalmas eszk\u00f6z, de jelent\u0151s h\u00e1tr\u00e1nya a v\u00e1r\u00f3 \u00fczenetek mem\u00f3ria-alap\u00fa t\u00e1rol\u00e1sa a alkalmaz\u00e1sban. Ez azt jelenti, hogy az alkalmaz\u00e1s \u00fajraind\u00edt\u00e1sa friss\u00edt\u00e9s vagy hiba miatt elvesz\u00edti az \u00f6sszes \u00fczenetet, amely a visszafejleszt\u00e9sre v\u00e1r. Mivel ez a pont kritikus a rendszer\u00fcnk sz\u00e1m\u00e1ra, tov\u00e1bbi vizsg\u00e1lat n\u00e9lk\u00fcl elvetett\u00fck.<\/p>\n

<\/p>\n

A spring-kafka maga t\u00f6bbf\u00e9le ContainerAwareErrorHandler implement\u00e1ci\u00f3t k\u00edn\u00e1l, p\u00e9ld\u00e1ul SeekToCurrentErrorHandler<\/a><\/noindex>, amellyel a hiba el\u0151fordul\u00e1sa eset\u00e9n az offset elmozd\u00edt\u00e1sa n\u00e9lk\u00fcl k\u00e9s\u0151bb kezelhetj\u00fck az \u00fczenetet. A spring-kafka 2.3 verzi\u00f3j\u00e1t\u00f3l lehet\u0151s\u00e9g van a BackOffPolicy be\u00e1ll\u00edt\u00e1s\u00e1ra is.<\/p>\n

<\/p>\n

Ez a megk\u00f6zel\u00edt\u00e9s lehet\u0151v\u00e9 teszi, hogy a visszafejlesztett \u00fczenetek t\u00fal\u00e9lj\u00e9k az alkalmaz\u00e1s \u00fajraind\u00edt\u00e1s\u00e1t, de a DLQ mechanizmus tov\u00e1bbra is hi\u00e1nyzik. Ezt a lehet\u0151s\u00e9get v\u00e1lasztottuk 2019 elej\u00e9n, optimist\u00e1n abban b\u00edzva, hogy a DLQ-ra nincs sz\u00fcks\u00e9g (szerencs\u00e9nk volt, \u00e9s val\u00f3ban nem volt sz\u00fcks\u00e9g r\u00e1 az alkalmaz\u00e1s p\u00e1r h\u00f3napj\u00e1n kereszt\u00fcl \u00fczemeltet\u00e9si rendszer\u00e9vel). A tempor\u00e1lis hib\u00e1k aktiv\u00e1lt\u00e1k a SeekToCurrentErrorHandler-t. A t\u00f6bbi hiba a napl\u00f3ban jelent meg, offset elmozdul\u00e1shoz vezetett, \u00e9s a feldolgoz\u00e1s a k\u00f6vetkez\u0151 \u00fczenetn\u00e9l folytat\u00f3dott.<\/p>\n

<\/p>\n

A v\u00e9gs\u0151 d\u00f6nt\u00e9s<\/h2>\n

<\/p>\n

A SeekToCurrentErrorHandler alap\u00fa megval\u00f3s\u00edt\u00e1s r\u00e1vett minket, hogy saj\u00e1t mechanizmust fejlessz\u00fcnk ki az \u00fczenetek \u00fajrak\u00fcld\u00e9s\u00e9re.<\/p>\n

<\/p>\n

Els\u0151sorban a megl\u00e9v\u0151 tapasztalatok kihaszn\u00e1l\u00e1s\u00e1ra \u00e9s az alkalmaz\u00e1s logik\u00e1j\u00e1t\u00f3l f\u00fcgg\u0151en a b\u0151v\u00edt\u00e9s\u00e9re t\u00f6rekedt\u00fcnk. Line\u00e1ris logik\u00e1j\u00fa alkalmaz\u00e1s eset\u00e9n optim\u00e1lis lett volna, ha a hib\u00e1s \u00fczenetek \u00fajraolvas\u00e1s\u00e1t egy r\u00f6vid id\u0151re le\u00e1ll\u00edtjuk, amelyet a visszah\u00edv\u00e1si strat\u00e9gia keretein bel\u00fcl hat\u00e1roztunk meg. M\u00e1s alkalmaz\u00e1sok eset\u00e9n szeretn\u00e9nk egy egys\u00e9ges pontot, amely biztos\u00edtja a visszah\u00edv\u00e1si strat\u00e9gia v\u00e9grehajt\u00e1s\u00e1t. Ezen k\u00edv\u00fcl ennek az egys\u00e9ges pontnak DLQ funkcionalit\u00e1ssal is rendelkeznie kell mindk\u00e9t megk\u00f6zel\u00edt\u00e9shez.<\/p>\n

<\/p>\n

A visszah\u00edv\u00e1si strat\u00e9gia maga az alkalmaz\u00e1sban kell, hogy t\u00e1rol\u00f3djon, amely felel\u0151s a k\u00f6vetkez\u0151 id\u0151intervallum megszerz\u00e9s\u00e9\u00e9rt, amikor ideiglenes hiba l\u00e9p fel.<\/p>\n

<\/p>\n

A Consumer meg\u00e1ll\u00edt\u00e1sa line\u00e1ris logik\u00e1j\u00fa alkalmaz\u00e1s eset\u00e9n<\/h3>\n

<\/p>\n

A spring-kafka haszn\u00e1lata sor\u00e1n a Consumer le\u00e1ll\u00edt\u00e1s\u00e1hoz a k\u00f3d nagyj\u00e1b\u00f3l \u00edgy n\u00e9zhet ki:<\/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        \/\/ to DLQ\n    }<\/code><\/pre>\n
<\/p>\n
A p\u00e9ld\u00e1ban a retryAt az az id\u0151, amikor \u00fajra kell ind\u00edtani a MessageListenerContainer-t, ha m\u00e9g mindig m\u0171k\u00f6dik. Az \u00fajraind\u00edt\u00e1s egy k\u00fcl\u00f6n sz\u00e1lban t\u00f6rt\u00e9nik, amely a TaskScheduler-ben fut, amit a spring is biztos\u00edt. <\/p>\n
<\/p>\n
A retryAt \u00e9rt\u00e9k\u00e9t a k\u00f6vetkez\u0151 m\u00f3don tal\u00e1ljuk meg:<\/p>\n
<\/p>\n
    \n
  1. Megkeress\u00fck a visszah\u00edv\u00e1sok sz\u00e1m\u00e1t.<\/li>\n
  2. A sz\u00e1ml\u00e1l\u00f3 \u00e9rt\u00e9ke alapj\u00e1n megkeress\u00fck a jelenlegi k\u00e9sleltet\u00e9si intervallumot a visszah\u00edv\u00e1si strat\u00e9gi\u00e1ban. A strat\u00e9gia mag\u00e1ban az alkalmaz\u00e1sban van deklar\u00e1lva, t\u00e1rol\u00e1s\u00e1ra JSON form\u00e1tumot v\u00e1lasztottunk.<\/li>\n
  3. A JSON t\u00f6mbben tal\u00e1lt intervallum tartalmazza azokat a m\u00e1sodperceket, amelyek eltelt\u00e9vel \u00fajra kell kezdeni a feldolgoz\u00e1st. Ez a m\u00e1sodperc sz\u00e1m hozz\u00e1ad\u00f3dik a jelenlegi id\u0151h\u00f6z, \u00edgy l\u00e9trehozva a retryAt \u00e9rt\u00e9k\u00e9t.<\/li>\n
  4. Ha az intervallum nem tal\u00e1lhat\u00f3, a retryAt \u00e9rt\u00e9ke null lesz, \u00e9s az \u00fczenet a DLQ-ba ker\u00fcl a k\u00e9zi feldolgoz\u00e1s \u00e9rdek\u00e9ben.<\/li>\n<\/ol>\n
    <\/p>\n
    Ezzel a megk\u00f6zel\u00edt\u00e9ssel csak a jelenleg feldolgoz\u00e1s alatt \u00e1ll\u00f3 \u00fczenetek sz\u00e1m\u00e1nak meg\u0151rz\u00e9se marad, p\u00e9ld\u00e1ul az alkalmaz\u00e1s mem\u00f3ri\u00e1j\u00e1ban. Az \u00fajrapr\u00f3b\u00e1lkoz\u00e1si sz\u00e1ml\u00e1l\u00f3 mem\u00f3ri\u00e1ban val\u00f3 t\u00e1rol\u00e1sa nem kritikus ehhez a megk\u00f6zel\u00edt\u00e9shez, mivel a line\u00e1ris logik\u00e1val rendelkez\u0151 alkalmaz\u00e1s nem tudja az eg\u00e9sz feldolgoz\u00e1st elv\u00e9gezni. A spring-retryt\u0151l elt\u00e9r\u0151en az alkalmaz\u00e1s \u00fajraind\u00edt\u00e1sa nem vezet az \u00f6sszes \u00fczenet elveszt\u00e9s\u00e9hez az \u00fajrafeldolgoz\u00e1shoz, hanem csup\u00e1n a strat\u00e9gia \u00fajraind\u00edt\u00e1s\u00e1hoz. <\/p>\n
    <\/p>\n
    Ez a megk\u00f6zel\u00edt\u00e9s seg\u00edt cs\u00f6kkenteni a k\u00fcls\u0151 rendszer terhel\u00e9s\u00e9t, amely esetleg nem el\u00e9rhet\u0151 nagyon nagy terhel\u00e9s miatt. M\u00e1s szavakkal, az \u00fajrafeldolgoz\u00e1s mellett a mint\u00e1t is siker\u00fclt megval\u00f3s\u00edtanunk. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    A mi eset\u00fcnkben a hibak\u00fcsz\u00f6b mind\u00f6ssze 1, \u00e9s a rendszer \u00e1ll\u00e1sid\u0151j\u00e9nek minimaliz\u00e1l\u00e1sa \u00e9rdek\u00e9ben, a h\u00e1l\u00f3zati fennakad\u00e1sok ideiglenes volta miatt, nagyon granularis \u00fajrapr\u00f3b\u00e1lkoz\u00e1si strat\u00e9gi\u00e1t alkalmazunk, r\u00f6vid k\u00e9sleltet\u00e9si id\u0151kkel. Ez nem biztos, hogy minden v\u00e1llalati alkalmaz\u00e1s sz\u00e1m\u00e1ra megfelel\u0151, ez\u00e9rt a hibak\u00fcsz\u00f6b \u00e9s a k\u00e9sleltet\u00e9si intervallum ar\u00e1ny\u00e1t a rendszer saj\u00e1toss\u00e1gainak figyelembev\u00e9tel\u00e9vel kell meghat\u00e1rozni.<\/p>\n
    <\/p>\n
    K\u00fcl\u00f6n alkalmaz\u00e1s az indeterminisztikus logik\u00e1j\u00fa alkalmaz\u00e1sok \u00fczeneteinek feldolgoz\u00e1s\u00e1ra.<\/h3>\n
    <\/p>\n
    Itt van egy p\u00e9lda k\u00f3d, amely \u00fczenetet k\u00fcld egy olyan alkalmaz\u00e1sba (Retryer), amely \u00fajra megpr\u00f3b\u00e1lja elk\u00fcldeni a DESTINATION t\u00e9m\u00e1ba az RETRY_AT id\u0151 el\u00e9r\u00e9sekor:<\/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
    A p\u00e9lda alapj\u00e1n l\u00e1that\u00f3, hogy sok inform\u00e1ci\u00f3t tov\u00e1bb\u00edtunk a fejl\u00e9cben. A RETRY_AT \u00e9rt\u00e9ke ugyanott van, mint a fogyaszt\u00f3 le\u00e1ll\u00edt\u00e1s\u00e1n kereszt\u00fcl t\u00f6rt\u00e9n\u0151 \u00fajrapr\u00f3b\u00e1lkoz\u00e1s mechanizmus\u00e1n\u00e1l. A DESTINATION \u00e9s RETRY_AT mellett tov\u00e1bb\u00edtjuk:<\/p>\n
    <\/p>\n