🥇Бозкоркунии рӯйдодҳои аз Кафка гирифташуда

Салом, Хабр.

Ба наздикй И аз тачрибаи худ накл кард дар бораи кадом параметрҳое, ки мо ҳамчун як даста бештар барои Кафка Истеҳсолкунанда ва Истеъмолкунанда истифода мебарем, то ба таҳвили кафолатнок наздиктар шавем. Дар ин мақола ман мехоҳам ба шумо бигӯям, ки чӣ гуна мо коркарди такрории ҳодисаро аз Кафка дар натиҷаи муваққатан дастрас набудани системаи беруна ташкил кардем.

Барномаҳои муосир дар муҳити хеле мураккаб кор мекунанд. Мантиқи тиҷорӣ дар стеки технологияи муосир печонида шудааст, ки дар тасвири Docker идорашаванда аз ҷониби оркестр ба монанди Kubernetes ё OpenShift кор мекунад ва бо дигар барномаҳо ё ҳалли корхонаҳо тавассути занҷири роутерҳои ҷисмонӣ ва виртуалӣ муошират мекунад. Дар чунин муҳит, чизе ҳамеша метавонад шикаста шавад, аз ин рӯ коркарди рӯйдодҳо, агар яке аз системаҳои беруна дастрас набошад, қисми муҳими равандҳои тиҷоратии мост.

Пеш аз Кафка чӣ гуна буд

Қаблан дар лоиҳа мо IBM MQ-ро барои интиқоли паёмҳои асинхронӣ истифода мебурдем. Агар дар давоми кори хадамот ягон хатогӣ рух диҳад, паёми қабулшударо метавон дар навбати мурда-нома (DLQ) барои таҳлили минбаъдаи дастӣ ҷойгир кард. DLQ дар паҳлӯи навбати воридот сохта шуда буд, паём дар дохили IBM MQ интиқол дода шуд.

Агар хато муваққатӣ бошад ва мо онро муайян карда тавонистем (масалан, ResourceAccessException дар занги HTTP ё MongoTimeoutException дар дархости MongoDb), пас стратегияи такрорӣ эътибор пайдо мекунад. Новобаста аз мантиқи шохабандии барнома, паёми аслӣ ё ба навбати система барои ирсоли таъхир ё ба замимаи алоҳидае, ки барои аз нав фиристодани паёмҳо кайҳо пеш сохта шуда буд, интиқол дода шуд. Ин рақами дубора фиристодани сарлавҳаи паёмро дар бар мегирад, ки ба фосилаи таъхир ё охири стратегияи сатҳи барнома алоқаманд аст. Агар мо ба охири стратегия расида бошем, вале системаи беруна ҳанӯз дастнорас бошад, пас паём дар DLQ барои таҳлили дастӣ ҷойгир карда мешавад.

Ҷустуҷӯи ҳалли

Ҷустуҷӯ дар Интернет, шумо метавонед зеринро пайдо кунед қарор қабул мекунад. Хулоса, пешниҳод карда мешавад, ки барои ҳар як фосилаи таъхир мавзӯъ эҷод карда, дар паҳлӯ татбиқ кардани барномаҳои истеъмолӣ, ки паёмҳоро бо таъхири зарурӣ мехонанд.

Сарфи назар аз шумораи зиёди баррасиҳои мусбӣ, ба назарам он комилан муваффақ нест. Пеш аз ҳама, зеро таҳиякунанда, ба ғайр аз иҷрои талаботи тиҷорат, бояд барои татбиқи механизми тавсифшуда вақти зиёд сарф кунад.

Илова бар ин, агар назорати дастрасӣ дар кластери Кафка фаъол бошад, шумо бояд барои эҷоди мавзӯъҳо ва таъмини дастрасии зарурӣ ба онҳо чанд вақт сарф кунед. Илова бар ин, ба шумо лозим меояд, ки параметри дурусти retention.ms-ро барои ҳар як мавзӯъҳои такрорӣ интихоб кунед, то паёмҳо вақти дубора фиристодан дошта бошанд ва аз он нопадид нашаванд. Татбиқ ва дархости дастрасӣ бояд барои ҳар як хидмати мавҷуда ё нав такрор карда шавад.

Биёед ҳоло бубинем, ки чӣ гуна механизмҳо умуман баҳор ва махсусан баҳор-кафкаро барои коркарди паёмҳо таъмин мекунанд. Spring-kafka аз баҳори такрорӣ вобастагии гузаранда дорад, ки абстраксияҳоро барои идоракунии BackOffPolicies гуногун таъмин мекунад. Ин як воситаи хеле чандир аст, аммо камбудии назарраси он нигоҳ доштани паёмҳо барои фиристодани дубора дар хотираи барнома мебошад. Ин маънои онро дорад, ки бозоғозкунии барнома бо сабаби навсозӣ ё хатои амалиётӣ боиси гум шудани ҳама паёмҳои мунтазири коркард мегардад. Азбаски ин нукта барои системаи мо муҳим аст, мо онро минбаъд баррасӣ накардем.

Масалан, худи spring-kafka якчанд амалисозии ContainerAwareErrorHandler-ро таъмин мекунад SeekToCurrentErrorHandler, ки бо он шумо метавонед паёмро дертар бидуни тағир додани ҷуброн ҳангоми хато коркард кунед. Аз версияи spring-kafka 2.3 сар карда, насб кардани BackOffPolicy имконпазир шуд.

Ин равиш ба паёмҳои коркардшуда имкон медиҳад, ки ҳангоми бозоғозии барнома зинда монад, аммо то ҳол механизми DLQ вуҷуд надорад. Мо ин вариантро дар аввали соли 2019 интихоб кардем, бо боварии некбинона, ки DLQ лозим нест (мо хушбахт будем ва воқеан пас аз чанд моҳи кор кардани барнома бо чунин системаи коркарди дубора ба он ниёз надоштем). Хатогиҳои муваққатӣ боиси оташ задани SeekToCurrentErrorHandler шуданд. Хатогиҳои боқимонда дар гузориш чоп карда шуданд, ки дар натиҷа офсет шуд ва коркард бо паёми навбатӣ идома ёфт.

Қарори ниҳоӣ

Татбиқ дар асоси SeekToCurrentErrorHandler моро водор сохт, ки механизми шахсии худро барои фиристодани паёмҳо таҳия кунем.

Пеш аз ҳама, мо мехостем, ки таҷрибаи мавҷударо истифода барем ва онро вобаста ба мантиқи барномаҳо васеъ кунем. Барои барномаи мантиқии хатӣ, қатъ кардани хондани паёмҳои нав дар муддати кӯтоҳе, ки стратегияи такрорӣ муайян кардааст, беҳтарин мебуд. Барои барномаҳои дигар, ман мехостам як нуқтаи ягона дошта бошам, ки стратегияи такрориро амалӣ кунад. Илова бар ин, ин нуқтаи ягона барои ҳарду равиш бояд функсияи DLQ дошта бошад.

Худи стратегияи такрорӣ бояд дар барнома нигоҳ дошта шавад, ки он барои дарёфти фосилаи навбатӣ ҳангоми рух додани хатогии муваққатӣ масъул аст.

Қатъи истеъмолкунанда барои барномаи мантиқии хатӣ

Ҳангоми кор бо Spring-kafka, рамзи боздоштани Истеъмолкунанда метавонад чунин бошад:

public void pauseListenerContainer(MessageListenerContainer listenerContainer, 
                                   Instant retryAt) {
        if (nonNull(retryAt) && listenerContainer.isRunning()) {
            listenerContainer.stop();
            taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
            return;
        }
        // to DLQ
    }

Дар мисол, retryAt вақти аз нав оғоз кардани MessageListenerContainer аст, агар он ҳоло ҳам кор кунад. Бозсозӣ дар як риштаи алоҳида дар TaskScheduler оғоз мешавад, ки татбиқи он низ то баҳор таъмин карда мешавад.

Мо арзиши retryAt-ро бо роҳи зерин пайдо мекунем:

Қимати ҳисобкунаки такрорӣ ба назар гирифта мешавад.
Дар асоси арзиши ҳисобкунак, фосилаи таъхири ҷорӣ дар стратегияи такрорӣ ҷустуҷӯ карда мешавад. Стратегия дар худи барнома эълон шудааст; мо барои нигоҳ доштани он формати JSON-ро интихоб кардем.
Фосилаи дар массиви JSON пайдошуда миқдори сонияҳоро дар бар мегирад, ки пас аз он коркард бояд такрор шавад. Ин шумораи сонияҳо ба вақти ҷорӣ илова карда мешавад, то арзиши retryAtро ташкил диҳад.
Агар фосила ёфт нашавад, он гоҳ арзиши retryAt нул аст ва паём ба DLQ барои таҳлили дастӣ фиристода мешавад.

Бо ин равиш, танҳо нигоҳ доштани шумораи зангҳои такрорӣ барои ҳар як паёме, ки ҳоло коркард мешавад, боқӣ мемонад, масалан дар хотираи барнома. Нигоҳ доштани ҳисоби такрорӣ дар хотира барои ин равиш муҳим нест, зеро барномаи мантиқии хатӣ коркардро дар маҷмӯъ идора карда наметавонад. Баръакси баҳори такрорӣ, бозоғозии барнома боиси гум шудани ҳамаи паёмҳо мегардад, ки дубора коркард карда шаванд, балки танҳо стратегияро аз нав оғоз мекунад.

Ин равиш кӯмак мекунад, ки сарбориро аз системаи беруна бардорад, ки аз сабаби бори хеле вазнин дастнорас аст. Бо ибораи дигар гуем, мо дар баробари аз нав кор кардан ба ичрои накша муваффак шудем сими барқ.

Дар ҳолати мо, ҳадди ақалли хатогӣ танҳо 1 аст ва барои кам кардани вақти бекории система аз сабаби қатъи муваққатии шабака, мо стратегияи хеле дақиқи такрориро бо фосилаҳои хурди таъхир истифода мебарем. Ин метавонад барои ҳама барномаҳои гурӯҳӣ мувофиқ набошад, бинобар ин муносибати байни ҳадди хато ва арзиши фосила бояд дар асоси хусусиятҳои система интихоб карда шавад.

Замимаи алоҳида барои коркарди паёмҳо аз замимаҳо бо мантиқи ғайримуқаррарӣ

Ин аст мисоли коде, ки ба чунин барнома паём мефиристад (Retryer), ки ҳангоми расидан ба вақти RETRY_AT ба мавзӯи DESTINATION дубора ирсол мекунад:


public <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, 
                         Instant retryAt, String counter, String groupId, Exception e) {
        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
        List<Header> arrayOfHeaders = 
            new ArrayList<>(Arrays.asList(headers.toArray()));
        updateHeader(arrayOfHeaders, GROUP_ID, groupId::getBytes);
        updateHeader(arrayOfHeaders, DESTINATION, retryToTopic::getBytes);
        updateHeader(arrayOfHeaders, ORIGINAL_PARTITION, 
                     () -> Integer.toString(record.partition()).getBytes());
        if (nonNull(retryAt)) {
            updateHeader(arrayOfHeaders, COUNTER, counter::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "retry"::getBytes);
            updateHeader(arrayOfHeaders, RETRY_AT, retryAt.toString()::getBytes);
        } else {
            updateHeader(arrayOfHeaders, REASON, 
                         ExceptionUtils.getStackTrace(e)::getBytes);
            updateHeader(arrayOfHeaders, SEND_TO, "backout"::getBytes);
        }
        ProducerRecord<K, V> messageToSend =
            new ProducerRecord<>(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
        kafkaTemplate.send(messageToSend);
    }

Мисол нишон медиҳад, ки бисёр маълумот дар сарлавҳаҳо интиқол дода мешавад. Қимати RETRY_AT ҳамон тавре, ки механизми кӯшиши такрорӣ тавассути таваққуфгоҳи Истеъмолкунанда пайдо мешавад. Илова ба DESTINATION ва RETRY_AT мо мегузарем:

GROUP_ID, ки тавассути он мо паёмҳоро барои таҳлили дастӣ ва ҷустуҷӯи содда гурӯҳбандӣ мекунем.
ORIGINAL_PARTITION барои кӯшиши нигоҳ доштани ҳамон истеъмолкунанда барои коркарди дубора. Ин параметр метавонад нул бошад, дар ин ҳолат қисмати нав бо истифода аз калиди record.key() паёми аслӣ ба даст оварда мешавад.
Қимати COUNTER барои риояи стратегияи такрорӣ навсозӣ карда шуд.
SEND_TO доимӣ аст, ки оё паём ҳангоми расидан ба RETRY_AT барои коркард фиристода мешавад ё дар DLQ ҷойгир шудааст.
САБАБ - сабаби қатъ шудани коркарди паём.

Retryer паёмҳоро барои аз нав фиристодан ва таҳлили дастӣ дар PostgreSQL нигоҳ медорад. Таймер вазифаеро оғоз мекунад, ки паёмҳоро бо RETRY_AT пайдо мекунад ва онҳоро ба қисмати ORIGINAL_PARTITION мавзӯи DESTINATION бо калиди record.key() мефиристад.

Пас аз фиристодан, паёмҳо аз PostgreSQL нест карда мешаванд. Таҳлили дастии паёмҳо дар UI оддӣ сурат мегирад, ки бо Retryer тавассути REST API ҳамкорӣ мекунад. Хусусиятҳои асосии он фиристодан ё нест кардани паёмҳо аз DLQ, дидани маълумоти хато ва ҷустуҷӯи паёмҳо, масалан аз рӯи номи хато мебошанд.

Азбаски назорати дастрасӣ дар кластерҳои мо фаъол аст, зарур аст, ки ба таври иловагӣ дастрасӣ ба мавзӯъеро, ки Retryer гӯш мекунад, дархост кунад ва ба Retryer иҷозат диҳад, ки ба мавзӯи DESTINATION нависад. Ин номувофиқ аст, аммо бар хилофи равиши мавзӯи фосилавӣ, мо барои идоракунии он як DLQ ва UI-и мукаммал дорем.

Ҳолатҳое мавҷуданд, ки мавзӯи воридшударо якчанд гурӯҳҳои гуногуни истеъмолкунандагон мехонанд, ки замимаҳои онҳо мантиқи гуногунро амалӣ мекунанд. Коркарди паём тавассути Retryer барои яке аз ин барномаҳо боиси такрори дигараш мегардад. Барои муҳофизат аз ин, мо барои коркарди дубора мавзӯи алоҳида эҷод мекунем. Мавзӯҳои воридотӣ ва такрорӣ метавонанд аз ҷониби ҳамон истеъмолкунанда бидуни маҳдудият хонда шаванд.

Бо нобаёнӣ, ин равиш функсияи ноқилро таъмин намекунад, аммо онро метавон бо истифода аз барнома илова кард баҳор-булут-netflix ё нав барқгирандаи абрии баҳорӣ, печонидани ҷойҳое, ки хидматҳои беруна ба абстраксияҳои мувофиқ даъват карда мешаванд. Илова бар ин, он имконпазир мегардад, ки ба интихоби стратегияи барои қалъа намуна, ки он низ метавонад муфид бошад. Масалан, дар spring-cloud-netflix ин метавонад ҳавзи ришта ё семафор бошад.

хулоса

Дар натиҷа, мо як барномаи алоҳида дорем, ки ба мо имкон медиҳад коркарди паёмҳоро такрор кунем, агар ягон системаи беруна муваққатан дастнорас бошад.

Яке аз бартариҳои асосии барнома дар он аст, ки онро системаҳои беруна, ки дар ҳамон кластери Кафка кор мекунанд, бидуни тағироти назаррас дар паҳлӯи онҳо истифода бурдан мумкин аст! Чунин барнома танҳо бояд ба мавзӯи такрорӣ дастрасӣ пайдо кунад, чанд сарлавҳаи Кафкаро пур кунад ва ба Retryer паём фиристад. Барои баланд бардоштани инфрасохтори иловагӣ лозим нест. Ва барои кам кардани шумораи паёмҳое, ки аз барнома ба Retryer ва бозгашт интиқол дода мешаванд, мо барномаҳоро бо мантиқи хатӣ муайян кардем ва онҳоро тавассути истгоҳи Истеъмол дубора коркард кардем.

Манбаъ: will.com

Коркарди рӯйдодҳои аз Кафка гирифташуда