Ciao, Habr.
Recentemente ho su quali parametri utilizziamo più 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à di un sistema esterno.
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ò sempre rompersi, quindi il ri-processamento degli eventi nel caso di indisponibilità di uno dei sistemi esterni è una parte fondamentale dei nostri processi aziendali.
Com'era prima di Kafka
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.
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 è 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 è ancora non disponibile, il messaggio verrà collocato nella DLQ per l'analisi manuale.
Ricerca della soluzione
, si possono trovare le seguenti informazioni . 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.
Nonostante un gran numero di recensioni positive, mi sembra che non sia del tutto riuscito. Innanzitutto perché, oltre a soddisfare le esigenze aziendali, lo sviluppatore dovrà dedicare molto tempo all'implementazione di meccanismi descritti.
Inoltre, se nel cluster Kafka è attivata la gestione degli accessi, sarà necessario investire del tempo per creare i topic e garantire i necessari permessi. Inoltre, sarà necessario trovare il giusto parametro retention.ms per ciascun topic di retry, affinché i messaggi possano essere reinviati senza scomparire. L'implementazione e la richiesta di accesso dovranno essere ripetute per ogni servizio esistente o nuovo.
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 è la conservazione dei messaggi in attesa di reinvio in memoria dell'applicazione. Ciò significa che il riavvio dell'applicazione a causa di un aggiornamento o di un errore durante il funzionamento comporterà la perdita di tutti i messaggi in attesa di rielaborazione. Poiché questo aspetto è critico per il nostro sistema, non abbiamo ritenuto opportuno considerarlo ulteriormente.
Spring-Kafka fornisce diverse implementazioni di ContainerAwareErrorHandler, ad esempio , 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, è possibile specificare una BackOffPolicy.
Questo approccio consente ai messaggi ri-processabili di sopravvivere al riavvio dell'applicazione, ma il meccanismo DLQ è ancora assente. Questa è 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 è 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.
Risultato finale
L'implementazione basata su SeekToCurrentErrorHandler ci ha spinti a sviluppare il nostro meccanismo per la reinvio dei messaggi.
Prima di tutto, volevamo utilizzare l'esperienza già 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à DLQ per entrambi gli approcci.
La strategia di ripetizione delle chiamate deve essere memorizzata nell'applicazione che si occupa di ottenere il successivo intervallo in caso di errore temporaneo.
Interruzione del Consumer per un'applicazione con logica lineare
Quando si lavora con spring-kafka, il codice per fermare il Consumer potrebbe apparire simile a questo:
public void pauseListenerContainer(MessageListenerContainer listenerContainer,
Instant retryAt) {
if (nonNull(retryAt) && listenerContainer.isRunning()) {
listenerContainer.stop();
taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
return;
}
// to DLQ
}Nell'esempio, retryAt è il momento in cui il MessageListenerContainer deve essere riavviato se è ancora attivo. Il riavvio avverrà in un thread separato avviato in TaskScheduler, la cui implementazione è fornita anch'essa da Spring.
Troviamo il valore di retryAt nel seguente modo:
- Viene cercato il valore del contatore dei tentativi.
- In base al valore del contatore, viene cercato l'attuale intervallo di ritardo nella strategia di retry. La strategia è dichiarata nell'applicazione stessa e per conservarla abbiamo scelto il formato JSON.
- 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.
- Se l'intervallo non viene trovato, il valore di retryAt è null e il messaggio verrà inviato alla DLQ per un'analisi manuale.
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 è critico per questo approccio, poiché un'applicazione con logica lineare non può elaborare nel suo insieme. A differenza di spring-retry, il riavvio dell'applicazione non comporterà la perdita di tutti i messaggi per un nuovo tentativo, ma semplicemente il riavvio della strategia.
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. .
Nel nostro caso, la soglia di errore è solo 1, e per minimizzare il tempo di inattività 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.
Un'applicazione dedicata per la gestione di messaggi da applicazioni con logica non deterministica
Ecco un esempio di codice che invia un messaggio a tale applicazione (Retryer), che ripeterà l'invio al tema DESTINATION quando viene raggiunto il tempo RETRY_AT:
public void retry(ConsumerRecord record, String retryToTopic,
Instant retryAt, String counter, String groupId, Exception e) {
Headers headers = ofNullable(record.headers()).orElse(new RecordHeaders());
List 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 messageToSend =
new ProducerRecord(retryTopic, null, null, record.key(), record.value(), arrayOfHeaders);
kafkaTemplate.send(messageToSend);
} Dall'esempio si evince che molte informazioni vengono trasmesse negli header. Il valore RETRY_AT è presente così come per il meccanismo di ripetizione tramite l'interruzione del Consumer. Oltre a DESTINATION e RETRY_AT, trasmettiamo:
- GROUP_ID, che utilizziamo per raggruppare i messaggi per l'analisi manuale e facilitare la ricerca.
- ORIGINAL_PARTITION, per cercare di mantenere lo stesso Consumer per il ri-processing. Questo parametro può essere uguale a null; in tal caso, una nuova partition sarà ottenuta tramite la chiave record.key() del messaggio originale.
- Il valore aggiornato di COUNTER, per seguire la strategia di retry.
- SEND_TO — una costante che indica se inviare il messaggio per un nuovo tentativo al raggiungimento di RETRY_AT o inserirlo nella DLQ.
- REASON — il motivo per cui l'elaborazione del messaggio è stata interrotta.
Il Retryer conserva i messaggi per la reinvio e l'analisi manuale in PostgreSQL. Un task è avviato da un timer, che trova i messaggi con RETRY_AT scaduto e li ri-invia nella partition ORIGINAL_PARTITION del topic DESTINATION con la chiave record.key().
Una volta inviato, i messaggi vengono eliminati da PostgreSQL. L'analisi manuale dei messaggi avviene in un'interfaccia utente semplice, che interagisce con Retryer tramite REST API. Le sue principali funzionalità includono la retransmissione o l'eliminazione dei messaggi dalla DLQ, la visualizzazione delle informazioni sugli errori e la ricerca dei messaggi, ad esempio per nome dell'errore.
Poiché nei nostri cluster è attivata la gestione degli accessi, è necessario richiedere l'accesso al topic che ascolta Retryer e consentire a Retryer di scrivere nel topic DESTINATION. Questo è scomodo, ma, a differenza dell'approccio con il topic a intervallo, abbiamo una DLQ completa e un'interfaccia utente per gestirla.
Ci sono casi in cui un topic in entrata è letto da diversi gruppi di consumer, le cui applicazioni implementano logiche diverse. La rielaborazione di un messaggio tramite Retryer per una di queste applicazioni porterà a un duplicato nell'altra. Per proteggersi da ciò, creiamo un topic separato per la rielaborazione. Il topic in entrata e il topic di retry possono essere letti dallo stesso Consumer senza alcuna limitazione.
Per impostazione predefinita, questo approccio non fornisce la possibilità di un circuito di interruzione, tuttavia può essere aggiunto all'applicazione tramite o nuovo , avvolgendo i punti di chiamata ai servizi esterni nelle rispettive astrazioni. Inoltre, si presenta la possibilità di scegliere la strategia per pattern, che può essere utile. Ad esempio, in spring-cloud-netflix può essere un thread pool o un semaforo.
Risultato
Di conseguenza, abbiamo ottenuto un'applicazione separata che consente di ripetere l'elaborazione del messaggio in caso di temporanea indisponibilità di un sistema esterno.
Uno dei principali vantaggi dell'applicazione è che può essere utilizzata da sistemi esterni che operano sullo stesso cluster Kafka, senza significative modifiche da parte loro! Questa applicazione avrà solo bisogno di accedere al topic di retry, compilare alcuni header Kafka e inviare un messaggio al Retryer. Non è necessario sollevare alcuna infrastruttura aggiuntiva. E per ridurre il numero di messaggi trasferiti dall'applicazione al Retryer e viceversa, abbiamo isolato applicazioni con logica lineare e realizzato il loro ripristino tramite l'interruzione del Consumer.
Fonte: habr.com
