{"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 la mia esperienza<\/a><\/noindex> su quali parametri utilizziamo pi\u00f9 spesso nel nostro team per Kafka Producer e Consumer per garantire la consegna dei messaggi. In questo articolo voglio raccontare come abbiamo organizzato il ri-processamento di un evento ricevuto da Kafka a causa della temporanea indisponibilit\u00e0 di un sistema esterno.<\/p>\n

<\/p>\n

Le applicazioni moderne operano in un ambiente molto complesso. La logica di business, incapsulata in uno stack tecnologico moderno, funzionante in un'immagine Docker, gestita da un orchestratore come Kubernetes o OpenShift, e in comunicazione con altre applicazioni o soluzioni aziendali attraverso una rete di router fisici e virtuali. In un tale contesto, qualcosa pu\u00f2 sempre rompersi, quindi il ri-processamento degli eventi nel caso di indisponibilit\u00e0 di uno dei sistemi esterni \u00e8 una parte fondamentale dei nostri processi aziendali.<\/p>\n

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

Com'era prima di Kafka<\/h2>\n

<\/p>\n

In precedenza, nel progetto utilizzavamo IBM MQ per la consegna asincrona dei messaggi. Se si verificava un errore durante l'esecuzione del servizio, il messaggio ricevuto poteva essere collocato nella coda di dead-letter (DLQ) per ulteriori analisi manuali. La DLQ veniva creata accanto alla coda di ingresso e il trasferimento del messaggio avveniva all'interno di IBM MQ. <\/p>\n

<\/p>\n

Se l'errore era temporaneo e riuscivamo a determinarlo (ad esempio, ResourceAccessException durante una chiamata HTTP o MongoTimeoutException durante una query in MongoDb), entrava in gioco la strategia di retry. Indipendentemente dal ramo logico dell'applicazione, il messaggio originale veniva spostato o nella coda di sistema per l'invio ritardato o in un'applicazione separata, che era stata creata tempo fa per reinviare i messaggi. In questo caso, nell'intestazione del messaggio viene registrato il numero del retry, che \u00e8 legato a un intervallo di attesa o alla fine della strategia a livello di applicazione. Se raggiungiamo la fine della strategia ma il sistema esterno \u00e8 ancora non disponibile, il messaggio verr\u00e0 collocato nella DLQ per l'analisi manuale.<\/p>\n

<\/p>\n

Ricerca della soluzione<\/h2>\n

<\/p>\n

Cercando su internet<\/a><\/noindex>, si possono trovare le seguenti informazioni soluzione<\/a><\/noindex>. In breve, si propone di creare un topic per ciascun intervallo di ritardo e di implementare sul lato dell'applicazione dei consumer che possano leggere i messaggi con il ritardo necessario. <\/p>\n

<\/p>\n

\"Rielaborazione<\/p>\n

<\/p>\n

Nonostante un gran numero di recensioni positive, mi sembra che non sia del tutto riuscito. Innanzitutto perch\u00e9, oltre a soddisfare le esigenze aziendali, lo sviluppatore dovr\u00e0 dedicare molto tempo all'implementazione di meccanismi descritti.<\/p>\n

<\/p>\n

Inoltre, se nel cluster Kafka \u00e8 attivata la gestione degli accessi, sar\u00e0 necessario investire del tempo per creare i topic e garantire i necessari permessi. Inoltre, sar\u00e0 necessario trovare il giusto parametro retention.ms per ciascun topic di retry, affinch\u00e9 i messaggi possano essere reinviati senza scomparire. L'implementazione e la richiesta di accesso dovranno essere ripetute per ogni servizio esistente o nuovo.<\/p>\n

<\/p>\n

Diamo ora un'occhiata ai meccanismi di rielaborazione dei messaggi forniti da Spring in generale e da Spring-Kafka in particolare. Spring-Kafka ha una dipendenza transitiva da Spring-Retry, che offre astrazioni per gestire diverse BackOffPolicy. Si tratta di uno strumento molto flessibile, ma il suo principale svantaggio \u00e8 la conservazione dei messaggi in attesa di reinvio in memoria dell'applicazione. Ci\u00f2 significa che il riavvio dell'applicazione a causa di un aggiornamento o di un errore durante il funzionamento comporter\u00e0 la perdita di tutti i messaggi in attesa di rielaborazione. Poich\u00e9 questo aspetto \u00e8 critico per il nostro sistema, non abbiamo ritenuto opportuno considerarlo ulteriormente.<\/p>\n

<\/p>\n

Spring-Kafka fornisce diverse implementazioni di ContainerAwareErrorHandler, ad esempio SeekToCurrentErrorHandler<\/a><\/noindex>, che consente di rielaborare un messaggio in seguito senza spostare l'offset in caso di errore. A partire dalla versione 2.3 di Spring-Kafka, \u00e8 possibile specificare una BackOffPolicy.<\/p>\n

<\/p>\n

Questo approccio consente ai messaggi ri-processabili di sopravvivere al riavvio dell'applicazione, ma il meccanismo DLQ \u00e8 ancora assente. Questa \u00e8 stata l'opzione che abbiamo scelto all'inizio del 2019, ottimisticamente ritenendo che il DLQ non fosse necessario (siamo stati fortunati e in effetti non \u00e8 stato necessario per diversi mesi di utilizzo dell'applicazione con questo sistema di ri-processamento). Gli errori temporanei hanno attivato SeekToCurrentErrorHandler. Gli altri errori venivano stampati nel log, portando a uno spostamento dell'offset, e il trattamento continuava con il messaggio successivo.<\/p>\n

<\/p>\n

Risultato finale<\/h2>\n

<\/p>\n

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

<\/p>\n

Prima di tutto, volevamo utilizzare l'esperienza gi\u00e0 acquisita e ampliarla in base alla logica dell'applicazione. Per un'applicazione con logica lineare, sarebbe ottimale interrompere la lettura di nuovi messaggi per un breve intervallo di tempo definito nell'ambito della strategia di ripetizione delle chiamate. Per le altre applicazioni, desideravamo avere un punto unico che garantisse l'esecuzione della strategia di ripetizione delle chiamate. Inoltre, questo punto unico dovrebbe avere funzionalit\u00e0 DLQ per entrambi gli approcci.<\/p>\n

<\/p>\n

La strategia di ripetizione delle chiamate deve essere memorizzata nell'applicazione che si occupa di ottenere il successivo intervallo in caso di errore temporaneo.<\/p>\n

<\/p>\n

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

<\/p>\n

Quando si lavora con spring-kafka, il codice per fermare il Consumer potrebbe apparire simile a questo:<\/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
Nell'esempio, retryAt \u00e8 il momento in cui il MessageListenerContainer deve essere riavviato se \u00e8 ancora attivo. Il riavvio avverr\u00e0 in un thread separato avviato in TaskScheduler, la cui implementazione \u00e8 fornita anch'essa da Spring. <\/p>\n
<\/p>\n
Troviamo il valore di retryAt nel seguente modo:<\/p>\n
<\/p>\n
    \n
  1. Viene cercato il valore del contatore dei tentativi.<\/li>\n
  2. In base al valore del contatore, viene cercato l'attuale intervallo di ritardo nella strategia di retry. La strategia \u00e8 dichiarata nell'applicazione stessa e per conservarla abbiamo scelto il formato JSON.<\/li>\n
  3. L'intervallo trovato nell'array JSON contiene il numero di secondi dopo i quali il trattamento deve essere ripetuto. Questo numero di secondi viene aggiunto all'ora attuale, formando il valore per retryAt.<\/li>\n
  4. Se l'intervallo non viene trovato, il valore di retryAt \u00e8 null e il messaggio verr\u00e0 inviato alla DLQ per un'analisi manuale.<\/li>\n<\/ol>\n
    <\/p>\n
    Con questo approccio, resta solo da mantenere il numero di invocazioni ripetute per ogni messaggio attualmente in elaborazione, ad esempio nella memoria dell'applicazione. Mantenere il contatore dei tentativi in memoria non \u00e8 critico per questo approccio, poich\u00e9 un'applicazione con logica lineare non pu\u00f2 elaborare nel suo insieme. A differenza di spring-retry, il riavvio dell'applicazione non comporter\u00e0 la perdita di tutti i messaggi per un nuovo tentativo, ma semplicemente il riavvio della strategia. <\/p>\n
    <\/p>\n
    Questo approccio aiuta a ridurre il carico sul sistema esterno, che potrebbe non essere disponibile a causa di un elevato carico di lavoro. In altre parole, oltre alla riprocessazione, siamo riusciti a implementare il pattern. circuit breaker<\/a><\/noindex>.<\/p>\n
    <\/p>\n
    Nel nostro caso, la soglia di errore \u00e8 solo 1, e per minimizzare il tempo di inattivit\u00e0 del sistema a causa di un temporaneo errore di rete, utilizziamo una strategia di ripetizione delle invocazioni molto granulare con brevi intervalli di attesa. Questo potrebbe non essere adatto a tutte le applicazioni del gruppo, quindi il rapporto tra la soglia di errore e la dimensione dell'intervallo deve essere scelto in base alle caratteristiche del sistema.<\/p>\n
    <\/p>\n
    Un'applicazione dedicata per la gestione di messaggi da applicazioni con logica non deterministica<\/h3>\n
    <\/p>\n
    Ecco un esempio di codice che invia un messaggio a tale applicazione (Retryer), che ripeter\u00e0 l'invio al tema DESTINATION quando viene raggiunto il tempo RETRY_AT:<\/p>\n
    <\/p>\n
    \npublic <K, V> void retry(ConsumerRecord<K, V> record, String retryToTopic, \n                         Instant retryAt, String counter, String groupId, Exception e) {\n        Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());\n        List<Header> 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<K, V> 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 evince che molte informazioni vengono trasmesse negli header. Il valore RETRY_AT \u00e8 presente cos\u00ec come per il meccanismo di ripetizione tramite l'interruzione del Consumer. Oltre a DESTINATION e RETRY_AT, trasmettiamo:<\/p>\n
    <\/p>\n