{"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\/da\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Genoptagelse af begivenheder modtaget fra Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Genoptagelse<\/p>\n

<\/p>\n

Hej, Habr.<\/p>\n

<\/p>\n

For nylig delte jeg min erfaring<\/a><\/noindex> om hvilke parametre vi oftest bruger i teamet til Kafka Producer og Consumer for at n\u00e6rme os garanteret levering. I denne artikel vil jeg fort\u00e6lle, hvordan vi organiserede genoptagelsen af en begivenhed modtaget fra Kafka som f\u00f8lge af midlertidig utilg\u00e6ngelighed af et eksternt system.<\/p>\n

<\/p>\n

Moderne applikationer fungerer i et meget komplekst milj\u00f8. Forretningslogik pakket ind i en moderne teknologistak, der k\u00f8rer i en Docker-image, som styres af en orkestrator som Kubernetes eller OpenShift, og kommunikerer med andre applikationer eller enterprise-l\u00f8sninger gennem en k\u00e6de af fysiske og virtuelle routere. I et s\u00e5dant milj\u00f8 kan der altid opst\u00e5 problemer, derfor er genoptagelse af begivenheder i tilf\u00e6lde af utilg\u00e6ngelighed af et eksternt system en vigtig del af vores forretningsprocesser.<\/p>\n

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

Hvordan det var f\u00f8r Kafka<\/h2>\n

<\/p>\n

Tidligere i projektet brugte vi IBM MQ til asynkron levering af beskeder. Ved enhver fejl i serviceprocessen kunne den modtagne besked blive placeret i en dead-letter-queue (DLQ) for manuel behandling. DLQ blev oprettet ved siden af den indkommende k\u00f8, og flytningen af beskeden skete inden for IBM MQ. <\/p>\n

<\/p>\n

Hvis fejlen var midlertidig, og vi kunne identificere det (f.eks. ResourceAccessException ved HTTP-kald eller MongoTimeoutException ved foresp\u00f8rgsel til MongoDb), tr\u00e5dte strategien for genkaldelser i kraft. Uanset hvordan applikationens logik grenes, blev den oprindelige besked flyttet til enten en systemk\u00f8 for forsinket levering eller til en separat applikation, der engang blev lavet til genlevering af beskeder. Heri skrives nummeret p\u00e5 genleveringen i beskedens header, hvilket er knyttet til forsinkelsesintervallet eller slutningen af strategien p\u00e5 applikationsniveau. Hvis vi n\u00e5r slutningen af strategien, men det eksterne system stadig er utilg\u00e6ngeligt, placeres beskeden i DLQ til manuel behandling.<\/p>\n

<\/p>\n

S\u00f8gning efter l\u00f8sning<\/h2>\n

<\/p>\n

Ved at s\u00f8ge p\u00e5 internettet<\/a><\/noindex>, kan man finde f\u00f8lgende l\u00f8sning<\/a><\/noindex>. Hvis man skal opsummere, foresl\u00e5s det at oprette et emne for hver forsinkelse og implementere Consumers p\u00e5 applikationssiden, som vil l\u00e6se beskeder med den n\u00f8dvendige forsinkelse. <\/p>\n

<\/p>\n

\"Genoptagelse<\/p>\n

<\/p>\n

P\u00e5 trods af det store antal positive anmeldelser virker det ikke helt vellykket for mig. F\u00f8rst og fremmest fordi udvikleren, udover at implementere forretningskravene, vil skulle bruge meget tid p\u00e5 at implementere den beskrevne mekanisme.<\/p>\n

<\/p>\n

Derudover, hvis adgangskontrol er aktiveret p\u00e5 Kafka-klyngen, vil der g\u00e5 noget tid med at oprette emner og sikre de n\u00f8dvendige adgange til dem. Udover dette skal den rette parameter retention.ms v\u00e6lges for hver af retry-emnerne, s\u00e5 beskederne n\u00e5r at blive sendt igen og ikke g\u00e5r tabt. Implementeringen og anmodningen om adgange skal gentages for hver eksisterende eller ny service.<\/p>\n

<\/p>\n

Lad os nu se p\u00e5, hvilke mekanismer der tilbydes af spring til genbehandling af beskeder generelt og af spring-kafka specifikt. Spring-kafka har en transitive afh\u00e6ngighed til spring-retry, som tilbyder abstraktioner til at styre forskellige BackOffPolicy. Dette er et ret fleksibelt v\u00e6rkt\u00f8j, men dets betydelige ulempe er, at beskeder til gensendelse opbevares i applikationens hukommelse. Det betyder, at genstart af applikationen p\u00e5 grund af opdateringer eller fejl under drift vil f\u00f8re til tab af alle beskeder, der venter p\u00e5 genbehandling. Da dette punkt er kritisk for vores system, besluttede vi ikke at overveje det yderligere.<\/p>\n

<\/p>\n

Selve spring-kafka tilbyder flere implementeringer af ContainerAwareErrorHandler, for eksempel SeekToCurrentErrorHandler<\/a><\/noindex>, som g\u00f8r det muligt at behandle en besked senere uden at flytte offset i tilf\u00e6lde af en fejl. Fra version 2.3 af spring-kafka er det blevet muligt at angive BackOffPolicy.<\/p>\n

<\/p>\n

Denne tilgang lader genbehandlede beskeder overleve genstart af applikationen, men mekanismen DLQ mangler stadig. Det var netop denne l\u00f8sning, vi valgte i starten af 2019, optimistisk i troen p\u00e5, at DLQ ikke ville v\u00e6re n\u00f8dvendigt (vi var heldige, og det blev faktisk ikke n\u00f8dvendigt i flere m\u00e5neder med drift af applikationen med dette system til genbehandling). Midlertidige fejl f\u00f8rte til aktivering af SeekToCurrentErrorHandler. Andre fejl blev udskrevet i loggen, f\u00f8rte til flytning af offset, og behandlingen fortsatte med den n\u00e6ste besked.<\/p>\n

<\/p>\n

Den endelige l\u00f8sning<\/h2>\n

<\/p>\n

Implementeringen baseret p\u00e5 SeekToCurrentErrorHandler har tilskyndet os til at udvikle vores egen mekanisme til genudsendelse af meddelelser.<\/p>\n

<\/p>\n

F\u00f8rst og fremmest \u00f8nskede vi at udnytte eksisterende erfaring og udvide den afh\u00e6ngigt af applikationens logik. For en applikation med line\u00e6r logik ville det v\u00e6re optimalt at stoppe med at l\u00e6se nye meddelelser i en kort periode, fastsat i genudsendelsesstrategien. For de andre applikationer \u00f8nskede vi at have et enkelt punkt, der ville sikre udf\u00f8relsen af genudsendelsesstrategien. Derudover skulle dette enhedspunkt have funktionalitet til DLQ for begge tilgange.<\/p>\n

<\/p>\n

Selve genudsendelsesstrategien skal v\u00e6re gemt i applikationen, der er ansvarlig for at modtage det n\u00e6ste interval ved opst\u00e5en af en midlertidig fejl.<\/p>\n

<\/p>\n

Stop af Consumer'en for applikationen med line\u00e6r logik<\/h3>\n

<\/p>\n

N\u00e5r man arbejder med spring-kafka, kan koden til at stoppe Consumer'en se omtrent s\u00e5dan ud:<\/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        \/\/ til DLQ\n    }<\/code><\/pre>\n
<\/p>\n
I eksemplet er retryAt det tidspunkt, hvor MessageListenerContainer skal genstartes, hvis det stadig er i gang. Genstarten vil finde sted i en separat tr\u00e5d, der er startet i TaskScheduler, hvis implementering ogs\u00e5 leveres af spring. <\/p>\n
<\/p>\n
V\u00e6rdien af retryAt finder vi p\u00e5 f\u00f8lgende m\u00e5de:<\/p>\n
<\/p>\n
    \n
  1. V\u00e6rdien af genudsendelsest\u00e6lleren s\u00f8ges.<\/li>\n
  2. I overensstemmelse med v\u00e6rdien af t\u00e6lleren findes det aktuelle forsinkelsesinterval i genudsendelsesstrategien. Strategien erkl\u00e6res i selve applikationen; til opbevaring har vi valgt JSON-formatet.<\/li>\n
  3. Det fundne interval i JSON-arrayet indeholder antallet af sekunder, hvorefter behandlingen skal gentages. Dette antal sekunder l\u00e6gges til den aktuelle tid, hvilket danner v\u00e6rdien for retryAt.<\/li>\n
  4. Hvis intervallet ikke findes, er v\u00e6rdien af retryAt null, og meddelelsen vil blive sendt til DLQ til manuel gennemgang.<\/li>\n<\/ol>\n
    <\/p>\n
    Med denne tilgang er det kun muligt at gemme antallet af genkald for hver besked, der i \u00f8jeblikket er under behandling, for eksempel i applikationens hukommelse. At gemme antallet af fors\u00f8g i hukommelsen er ikke kritisk for denne tilgang, da en applikation med line\u00e6r logik ikke kan udf\u00f8re behandlingen som helhed. I mods\u00e6tning til spring-retry vil en genstart af applikationen ikke f\u00f8re til tab af alle beskeder til genbehandling, men blot genstarte strategien. <\/p>\n
    <\/p>\n
    Denne tilgang hj\u00e6lper med at reducere belastningen p\u00e5 det eksterne system, som muligvis ikke er tilg\u00e6ngeligt p\u00e5 grund af en meget h\u00f8j belastning. Med andre ord har vi udover genbehandling opn\u00e5et implementeringen af et m\u00f8nster. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    I vores tilf\u00e6lde er fejlt\u00e6rsklen kun 1, og for at minimere systemets nedetid p\u00e5 grund af midlertidige netv\u00e6rksproblemer anvender vi en meget granular strategi for genkald med sm\u00e5 ventetider. Dette er m\u00e5ske ikke egnet til alle applikationer i koncernen, s\u00e5 forholdet mellem fejlt\u00e6rsklen og st\u00f8rrelsen p\u00e5 intervallet skal tilpasses baseret p\u00e5 systemets egenskaber.<\/p>\n
    <\/p>\n
    En s\u00e6rskilt applikation til behandling af beskeder fra applikationer med nondeterministisk logik<\/h3>\n
    <\/p>\n
    Her er et kodeeksempel, der sender en besked til en s\u00e5dan applikation (Retryer), som vil fors\u00f8ge at sende igen til emnet DESTINATION, n\u00e5r tiden RETRY_AT er n\u00e5et:<\/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
    Eksemplet viser, at der overf\u00f8res meget information i headerne. V\u00e6rdien RETRY_AT findes ligeledes som for mekanismen til gentagelse via stoppet af Consumer\u2019en. Udover DESTINATION og RETRY_AT overf\u00f8rer vi:<\/p>\n
    <\/p>\n