Ciao, Habr.
Recentemente ho su quali parametri utilizziamo più 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à temporanea di un sistema esterno.
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ò sempre andare storto, quindi la re-elaborazione degli eventi in caso di inaccessibilità di uno dei sistemi esterni è una parte importante dei nostri processi aziendali.
Come era prima di Kafka
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.
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.
Ricerca della soluzione
, si può trovare quanto segue . 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.
Nonostante il grande numero di recensioni positive, mi sembra non del tutto riuscito. Prima di tutto perché, oltre a soddisfare i requisiti di business, lo sviluppatore dovrà spendere molto tempo per implementare il meccanismo descritto.
Inoltre, se il cluster Kafka ha attivata la gestione degli accessi, ci vorrà del tempo per creare i topic e garantire i necessari diritti di accesso. In aggiunta a questo, sarà necessario trovare il giusto parametro retention.ms per ciascuno dei topic di retry, affinché 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.
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 è uno strumento piuttosto flessibile, ma un suo notevole svantaggio è 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à alla perdita di tutti i messaggi in attesa di rielaborazione. Poiché questo punto è critico per il nostro sistema, non l'abbiamo considerato ulteriormente.
Spring-Kafka stesso fornisce diverse implementazioni di ContainerAwareErrorHandler, ad esempio , che permette di elaborare successivamente un messaggio senza spostare l'offset in caso di errore. A partire dalla versione 2.3 di Spring-Kafka è stata introdotta la possibilità di definire una BackOffPolicy.
Questo approccio consente ai messaggi rielaborati di sopravvivere al riavvio dell'applicazione, ma il meccanismo DLQ è ancora assente. È 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 è 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.
La soluzione finale
L'implementazione basata su SeekToCurrentErrorHandler ci ha spinto a sviluppare il nostro meccanismo per la reinvio dei messaggi.
In primo luogo, volevamo utilizzare l'esperienza già 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à DLQ per entrambi gli approcci.
La strategia di retry stessa dovrebbe essere conservata nell'applicazione che si occupa di ricevere il successivo intervallo in caso di errore temporaneo.
Arresto del Consumer per un'applicazione con logica lineare
Durante l'uso di spring-kafka, il codice per arrestare il Consumer può apparire più o meno così:
public void pauseListenerContainer(MessageListenerContainer listenerContainer,
Instant retryAt) {
if (nonNull(retryAt) && listenerContainer.isRunning()) {
listenerContainer.stop();
taskScheduler.schedule(() -> listenerContainer.start(), retryAt);
return;
}
// a DLQ
}Nell'esempio, retryAt è il momento in cui riavviare il MessageListenerContainer, se è ancora in esecuzione. Il riavvio avverrà in un thread separato avviato nel TaskScheduler, la cui implementazione è fornita anche da spring.
Troviamo il valore di retryAt nel modo seguente:
- Si cerca il valore del contatore dei retry.
- In base al valore del contatore, si cerca l'intervallo di attesa attuale nella strategia di retry. La strategia è dichiarata nell'applicazione stessa e per la sua memorizzazione abbiamo scelto il formato JSON.
- L'intervallo trovato nell'array JSON contiene il numero di secondi dopo i quali sarà necessario ripetere l'elaborazione. Questo numero di secondi viene aggiunto al tempo attuale, creando il valore per retryAt.
- Se l'intervallo non viene trovato, il valore di retryAt è null e il messaggio sarà inviato a DLQ per un'analisi manuale.
Con questo approccio, resta solo da conservare il numero di tentativi per ogni messaggio che è attualmente in trattamento, ad esempio nella memoria dell'applicazione. La conservazione del contatore dei tentativi in memoria non è critica per questo approccio, poiché un'applicazione con logica lineare non può eseguire la lavorazione nel suo complesso. A differenza di spring-retry, il riavvio dell'applicazione non comporterà la perdita di tutti i messaggi da rielaborare, ma semplicemente un riavvio della strategia.
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 .
Nel nostro caso, la soglia di errore è 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.
Un'applicazione separata per l'elaborazione di messaggi provenienti da applicazioni con logica non deterministica
Ecco un esempio di codice che invia un messaggio a tale applicazione (Retryer), che eseguirà un nuovo invio al topic DESTINATION quando raggiunge 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 può 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:
- GROUP_ID, utilizzato per raggruppare i messaggi per analisi manuale e semplificazione della ricerca.
- ORIGINAL_PARTITION, per cercare di mantenere lo stesso Consumer per un rielaborazione. Questo parametro può essere nullo, in tal caso una nuova partition sarà ottenuta tramite la chiave record.key() del messaggio originale.
- Valore aggiornato del COUNTER, per seguire la strategia delle ripetizioni.
- SEND_TO — costante che indica se il messaggio debba essere reinviato per la rielaborazione al raggiungimento di RETRY_AT o inserito in DLQ.
- REASON — motivo per cui l'elaborazione del messaggio è stata interrotta.
Il Retryer conserva i messaggi per il reinvio e l'analisi manuale in PostgreSQL. In base a un timer, viene avviato un compito che trova i messaggi con RETRY_AT raggiunto e li reinvia nella partition ORIGINAL_PARTITION del topic DESTINATION con la chiave record.key().
Dopo l'invio, i messaggi vengono rimossi da PostgreSQL. L'analisi manuale dei messaggi avviene tramite un'interfaccia utente semplice, che interagisce con il Retryer tramite REST API. Le sue principali caratteristiche includono il reinvio o la rimozione dei messaggi dalla DLQ, la visualizzazione delle informazioni sugli errori e la ricerca di messaggi, ad esempio per nome dell'errore.
Poiché nei nostri cluster è abilitata la gestione degli accessi, è necessario richiedere accessi aggiuntivi al topic a cui ascolta il Retryer e consentire a quest'ultimo di scrivere nel topic DESTINATION. È scomodo, ma a differenza dell'approccio con il topic a intervallo, abbiamo a disposizione una DLQ completa e un'interfaccia per la sua gestione.
Ci sono casi in cui il topic in ingresso viene letto da più gruppi di consumer diversi, le cui applicazioni implementano logiche diverse. La rielaborazione di un messaggio tramite il Retryer per una di queste applicazioni darà luogo a un duplicato nell'altra. Per proteggersi da questo, creiamo un topic separato per la rielaborazione. Il topic in ingresso e il topic di retry possono essere letti dallo stesso Consumer senza alcuna restrizione.
Di default, questo approccio non fornisce la possibilità di un circuito di interruzione, tuttavia può essere aggiunto all'applicazione tramite o il nuovo , avvolgendo i punti di chiamata ai servizi esterni nelle relative astrazioni. Inoltre, si offre la possibilità di scegliere una strategia per un modello, che può essere utile. Ad esempio, in spring-cloud-netflix può trattarsi di un thread pool o di un semaforo.
Conclusione
Di conseguenza, abbiamo creato un'applicazione separata che consente di ripetere l'elaborazione di un messaggio in caso di temporanea indisponibilità di un sistema esterno.
Uno dei principali vantaggi dell'applicazione è che possono utilizzarla sistemi esterni che operano nello stesso cluster Kafka, senza modifiche significative da parte loro! Questa applicazione avrà solo bisogno di accedere al topic di retry, riempiere alcuni header Kafka e inviare il messaggio al Retryer. Non è necessario creare infrastrutture aggiuntive. Inoltre, per ridurre il numero di messaggi trasferiti dall'applicazione al Retryer e viceversa, abbiamo isolato le applicazioni con logica lineare e implementato la ripetizione attraverso l'interruzione del Consumer.
Fonte: habr.com
