{"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\/it\/blog\/administrirovanie\/povtornaya-obrabotka-sobytij-poluchennyh-iz-kafka","title":{"rendered":"Rielaborazione degli eventi ricevuti da Kafka","gt_translate_keys":[{"key":"rendered","format":"text"}]},"content":{"rendered":"

\"Rielaborazione<\/p>\n

<\/p>\n

Ciao, Habr.<\/p>\n

<\/p>\n

Recentemente ho condiviso un'esperienza<\/a><\/noindex> su quali parametri utilizziamo pi\u00f9 frequentemente all'interno del nostro team per Kafka Producer e Consumer, al fine di avvicinarci alla consegna garantita. In questo articolo voglio raccontare come abbiamo organizzato la re-elaborazione di un evento ricevuto da Kafka, a seguito dell'inaccessibilit\u00e0 temporanea di un sistema esterno.<\/p>\n

<\/p>\n

Le applicazioni moderne operano in un ambiente molto complesso. La logica aziendale, incapsulata in un moderno stack tecnologico, funziona in un'immagine Docker gestita da un orchestratore come Kubernetes o OpenShift e comunica con altre applicazioni o soluzioni aziendali attraverso una rete di router fisici e virtuali. In un tale contesto, qualcosa pu\u00f2 sempre andare storto, quindi la re-elaborazione degli eventi in caso di inaccessibilit\u00e0 di uno dei sistemi esterni \u00e8 una parte importante dei nostri processi aziendali.<\/p>\n

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

Come era prima di Kafka<\/h2>\n

<\/p>\n

In precedenza, nel progetto utilizzavano IBM MQ per la consegna asincrona dei messaggi. Quando si verificava un errore nel processo di funzionamento del servizio, il messaggio ricevuto poteva essere collocato in una coda di messaggi non elaborabili (DLQ) per un successivo esame manuale. La DLQ veniva creata accanto alla coda in ingresso e il messaggio veniva spostato all'interno di IBM MQ. <\/p>\n

<\/p>\n

Se l'errore era di natura temporanea e riuscivamo a determinarlo (ad esempio, ResourceAccessException durante una chiamata HTTP o MongoTimeoutException durante una richiesta a MongoDb), entrava in gioco la strategia dei ripetuti tentativi. Indipendentemente dallo branching della logica dell'applicazione, il messaggio originale veniva spostato o in una coda di sistema per l'invio posticipato, oppure in un'applicazione separata creata tempo fa per il reinvio dei messaggi. In questo caso, nell'intestazione del messaggio veniva registrato il numero del tentativo di reinvio, legato a un intervallo di ritardo o alla fine della strategia a livello applicativo. Se raggiungevamo la fine della strategia ma il sistema esterno era ancora inaccessibile, il messaggio veniva collocato nella DLQ per un esame manuale.<\/p>\n

<\/p>\n

Ricerca della soluzione<\/h2>\n

<\/p>\n

Cercando su internet<\/a><\/noindex>, si pu\u00f2 trovare quanto segue risultato<\/a><\/noindex>. In breve, viene proposta la creazione di un argomento per ogni intervallo di ritardo e l'implementazione da parte dell'applicazione di Consumer che leggeranno i messaggi con il ritardo necessario. <\/p>\n

<\/p>\n

\"Rielaborazione<\/p>\n

<\/p>\n

Nonostante il grande numero di recensioni positive, mi sembra non del tutto riuscito. Prima di tutto perch\u00e9, oltre a soddisfare i requisiti di business, lo sviluppatore dovr\u00e0 spendere molto tempo per implementare il meccanismo descritto.<\/p>\n

<\/p>\n

Inoltre, se il cluster Kafka ha attivata la gestione degli accessi, ci vorr\u00e0 del tempo per creare i topic e garantire i necessari diritti di accesso. In aggiunta a questo, sar\u00e0 necessario trovare il giusto parametro retention.ms per ciascuno dei topic di retry, affinch\u00e9 i messaggi possano essere reinviati e non vengano persi. L'implementazione e la richiesta di accessi dovranno essere ripetute per ciascun servizio esistente o nuovo.<\/p>\n

<\/p>\n

Vediamo ora quali meccanismi per la rielaborazione dei messaggi ci offre Spring in generale e Spring-Kafka in particolare. Spring-Kafka ha una dipendenza transitiva su Spring-Retry, che fornisce astrazioni per gestire diverse BackOffPolicy. Questo \u00e8 uno strumento piuttosto flessibile, ma un suo notevole svantaggio \u00e8 l'archiviazione dei messaggi per il reinvio nella memoria dell'applicazione. Questo significa che un riavvio dell'applicazione a causa di un aggiornamento o di un errore durante l'utilizzo porter\u00e0 alla perdita di tutti i messaggi in attesa di rielaborazione. Poich\u00e9 questo punto \u00e8 critico per il nostro sistema, non l'abbiamo considerato ulteriormente.<\/p>\n

<\/p>\n

Spring-Kafka stesso fornisce diverse implementazioni di ContainerAwareErrorHandler, ad esempio SeekToCurrentErrorHandler<\/a><\/noindex>, che permette di elaborare successivamente un messaggio senza spostare l'offset in caso di errore. A partire dalla versione 2.3 di Spring-Kafka \u00e8 stata introdotta la possibilit\u00e0 di definire una BackOffPolicy.<\/p>\n

<\/p>\n

Questo approccio consente ai messaggi rielaborati di sopravvivere al riavvio dell'applicazione, ma il meccanismo DLQ \u00e8 ancora assente. \u00c8 proprio questa opzione che abbiamo scelto all'inizio del 2019, ottimisticamente ritenendo che il DLQ non sarebbe stato necessario (siamo stati fortunati e in effetti non \u00e8 stato necessario per diversi mesi di utilizzo dell'applicazione con questo sistema di rielaborazione). Gli errori temporanei portavano all'attivazione di SeekToCurrentErrorHandler. Altri errori venivano registrati nei log, portavano a uno spostamento dell'offset e l'elaborazione continuava con il messaggio successivo.<\/p>\n

<\/p>\n

La soluzione finale<\/h2>\n

<\/p>\n

L'implementazione basata su SeekToCurrentErrorHandler ci ha spinto a sviluppare il nostro meccanismo per la reinvio dei messaggi.<\/p>\n

<\/p>\n

In primo luogo, volevamo utilizzare l'esperienza gi\u00e0 esistente e ampliarla in base alla logica dell'applicazione. Per un'applicazione con logica lineare, sarebbe stato ottimale interrompere la lettura di nuovi messaggi per un breve intervallo di tempo, stabilito all'interno della strategia di retry. Per le altre applicazioni, volevamo avere un unico punto che garantisse l'esecuzione della strategia di retry. Inoltre, questo unico punto dovrebbe possedere funzionalit\u00e0 DLQ per entrambi gli approcci.<\/p>\n

<\/p>\n

La strategia di retry stessa dovrebbe essere conservata nell'applicazione che si occupa di ricevere il successivo intervallo in caso di errore temporaneo.<\/p>\n

<\/p>\n

Arresto del Consumer per un'applicazione con logica lineare<\/h3>\n

<\/p>\n

Durante l'uso di spring-kafka, il codice per arrestare il Consumer pu\u00f2 apparire pi\u00f9 o meno cos\u00ec:<\/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        \/\/ a DLQ\n    }<\/code><\/pre>\n
<\/p>\n
Nell'esempio, retryAt \u00e8 il momento in cui riavviare il MessageListenerContainer, se \u00e8 ancora in esecuzione. Il riavvio avverr\u00e0 in un thread separato avviato nel TaskScheduler, la cui implementazione \u00e8 fornita anche da spring. <\/p>\n
<\/p>\n
Troviamo il valore di retryAt nel modo seguente:<\/p>\n
<\/p>\n
    \n
  1. Si cerca il valore del contatore dei retry.<\/li>\n
  2. In base al valore del contatore, si cerca l'intervallo di attesa attuale nella strategia di retry. La strategia \u00e8 dichiarata nell'applicazione stessa e per la sua memorizzazione abbiamo scelto il formato JSON.<\/li>\n
  3. L'intervallo trovato nell'array JSON contiene il numero di secondi dopo i quali sar\u00e0 necessario ripetere l'elaborazione. Questo numero di secondi viene aggiunto al tempo attuale, creando il valore per retryAt.<\/li>\n
  4. Se l'intervallo non viene trovato, il valore di retryAt \u00e8 null e il messaggio sar\u00e0 inviato a DLQ per un'analisi manuale.<\/li>\n<\/ol>\n
    <\/p>\n
    Con questo approccio, resta solo da conservare il numero di tentativi per ogni messaggio che \u00e8 attualmente in trattamento, ad esempio nella memoria dell'applicazione. La conservazione del contatore dei tentativi in memoria non \u00e8 critica per questo approccio, poich\u00e9 un'applicazione con logica lineare non pu\u00f2 eseguire la lavorazione nel suo complesso. A differenza di spring-retry, il riavvio dell'applicazione non comporter\u00e0 la perdita di tutti i messaggi da rielaborare, ma semplicemente un riavvio della strategia. <\/p>\n
    <\/p>\n
    Questo approccio aiuta a ridurre il carico su un sistema esterno, che potrebbe non essere disponibile a causa di un carico molto alto. In altre parole, oltre alla rielaborazione, abbiamo realizzato l'implementazione del pattern circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    Nel nostro caso, la soglia di errore \u00e8 di solo 1, e per minimizzare il downtime del sistema a causa di un temporaneo guasto di rete, utilizziamo una strategia di tentativi molto granulare con brevi intervalli di attesa. Potrebbe non essere adatta a tutte le applicazioni del gruppo, quindi la relazione tra la soglia di errore e la dimensione dell'intervallo deve essere regolata in base alle caratteristiche del sistema.<\/p>\n
    <\/p>\n
    Un'applicazione separata per l'elaborazione di messaggi provenienti da applicazioni con logica non deterministica<\/h3>\n
    <\/p>\n
    Ecco un esempio di codice che invia un messaggio a tale applicazione (Retryer), che eseguir\u00e0 un nuovo invio al topic DESTINATION quando raggiunge il tempo 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
    Dall'esempio si pu\u00f2 notare che molte informazioni vengono trasmesse negli header. Il valore di RETRY_AT si trova allo stesso modo di quanto avviene per il meccanismo di ripetizione tramite la fermata del Consumer. Oltre a DESTINATION e RETRY_AT, trasmettiamo:<\/p>\n
    <\/p>\n