Ciao, Habr!
Lavoro nel team di Tinkoff che si occupa dello sviluppo del proprio centro di notifiche. Per lo più sviluppo in Java utilizzando Spring Boot e risolvo vari problemi tecnici che sorgono nel progetto.
La maggior parte dei nostri microservizi interagisce in modo asincrono tra loro tramite un broker di messaggi. In precedenza utilizzavamo IBM MQ come broker, ma ha smesso di poter reggere il carico, pur avendo elevate garanzie di consegna.
Come sostituzione ci è stato proposto Apache Kafka, che ha un alto potenziale di scalabilità, ma purtroppo richiede un approccio quasi personalizzato alla configurazione per diversi scenari. Inoltre, il meccanismo di consegna almeno una volta, che funziona in Kafka per impostazione predefinita, non permetteva di mantenere il livello di coerenza necessario di default. Di seguito condividerò la nostra esperienza di configurazione di Kafka, in particolare parlerò di come configurare e vivere con la consegna esattamente una volta.
Consegna garantita e non solo
Le impostazioni di cui parleremo aiuteranno a prevenire una serie di problemi con le impostazioni di connessione predefinite. Ma prima voglio mettere in evidenza un'impostazione che faciliterà un eventuale debug.
Questo sarà utile per client.id per il Producer e il Consumer. A prima vista, come valore si può utilizzare il nome dell'applicazione, e nella maggior parte dei casi funzionerà. Tuttavia, la situazione in cui un'applicazione utilizza più Consumer e si assegna lo stesso client.id porta al seguente avviso:
org.apache.kafka.common.utils.AppInfoParser — Errore nella registrazione dell'AppInfo mbean javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=kafka.test-0Se desideri utilizzare JMX in un'applicazione con Kafka, questo potrebbe essere un problema. In questo caso, è meglio utilizzare come valore del client.id una combinazione tra il nome dell'applicazione e, ad esempio, il nome del topic. Il risultato della nostra configurazione può essere visualizzato nell'output del comando kafka-consumer-groups delle utilità di Confluent:
Ora esaminiamo lo scenario della consegna garantita di un messaggio. Un Kafka Producer ha un'impostazione acks, che permette di configurare dopo quanti acknowledge il leader del cluster deve considerare il messaggio registrato con successo. Questa impostazione può assumere i seguenti valori:
- 0 — gli acknowledge non verranno conteggiati.
- 1 — impostazione predefinita, è richiesto l'acknowledge solo da 1 replica.
- −1 — sono necessari acknowledge da tutti i replica sincronizzati (configurazione del cluster min.insync.replicas).
Dai valori elencati si può vedere che un acks pari a −1 offre le garanzie più forti che il messaggio non venga perso.
Come sappiamo, i sistemi distribuiti non sono affidabili. Per proteggersi da guasti temporanei, il Kafka Producer offre il parametro retries, che consente di definire il numero di tentativi di reinvio entro delivery.timeout.ms. Poiché il parametro retries ha un valore predefinito di Integer.MAX_VALUE (2147483647), il numero di reinvii del messaggio può essere regolato cambiando solo delivery.timeout.ms.
Ci stiamo avvicinando alla consegna exactly once
Le impostazioni sopra menzionate consentono al nostro Producer di consegnare messaggi con elevata garanzia. Ora parliamo di come garantire la registrazione di una sola copia del messaggio nel topic di Kafka? Nel caso più semplice, per questo il Producer deve impostare il parametro enable.idempotence su true. L'idempotenza garantisce la registrazione di un solo messaggio in una specifica partizione di un topic. Le condizioni preliminari per abilitare l'idempotenza sono i valori acks = all, retry > 0, max.in.flight.requests.per.connection ≤ 5. Se questi parametri non sono specificati dallo sviluppatore, verranno automaticamente impostati i valori sopraindicati.
Quando l'idempotenza è configurata, è necessario assicurarsi che messaggi identici finiscano sempre nelle stesse partizioni. Questo può essere fatto configurando la chiave e il parametro partitioner.class sul Producer. Iniziamo dalla chiave. Per ogni invio, deve essere la stessa. Questo è facile da realizzare utilizzando un identificatore aziendale del messaggio originale. Il parametro partitioner.class ha come valore predefinito — . Con questa strategia di partizionamento predefinita, agiamo in questo modo:
- Se la partizione è esplicitamente specificata durante l'invio del messaggio, allora la utilizziamo.
- Se la partizione non è specificata, ma è stata fornita una chiave — scegliamo la partizione in base all'hash della chiave.
- Se né la partizione né la chiave sono specificate — scegliamo le partizioni in modo ciclico (round-robin).
Inoltre, l'uso della chiave e l'invio idempotente con il parametro max.in.flight.requests.per.connection = 1 ti offre un'elaborazione ordinata dei messaggi su Consumer. È importante ricordare che, se nel tuo cluster è configurato il controllo dell'accesso, avrai bisogno dei diritti per la scrittura idempotente nel topic.
Se ti manca la possibilità di inviare in modo idempotente per chiave o se la logica lato Producer richiede di mantenere la coerenza dei dati tra diverse partizioni, le transazioni possono aiutarti. Inoltre, tramite una transazione a catena, puoi sincronizzare condizionalmente una scrittura in Kafka, ad esempio, con una scrittura nel database. Per abilitare l'invio transazionale, il Producer deve essere idempotente e deve anche impostare transactional.id. Se nel tuo cluster Kafka è implementato il controllo dell'accesso, anche per la scrittura transazionale, così come per quella idempotente, saranno necessari diritti di scrittura, che possono essere forniti tramite una maschera utilizzando il valore memorizzato in transactional.id.
Formalmente, qualsiasi stringa può essere utilizzata come identificatore della transazione, ad esempio il nome dell'applicazione. Tuttavia, se stai eseguendo più istanze della stessa applicazione con lo stesso transactional.id, la prima istanza avviata verrà fermata con un errore, poiché Kafka la considererà un processo zombie.
org.apache.kafka.common.errors.ProducerFencedException: Il Producer ha tentato un'operazione con un'epoca precedente. O c'è un producer più recente con lo stesso transactionalId, o la transazione del producer è scaduta dal broker.Per risolvere questo problema, aggiungiamo al nome dell'applicazione un suffisso con il nome host ottenuto dalle variabili d'ambiente.
Il Producer è configurato, ma le transazioni su Kafka gestiscono solo l'ambito del messaggio. Indipendentemente dallo stato della transazione, il messaggio viene immediatamente inviato nel topic, ma possiede attributi sistemici aggiuntivi.
Affinché tali messaggi non vengano letti dal Consumer prima del tempo, è necessario impostare il parametro isolation.level su read_committed. Questo Consumer potrà leggere i messaggi non transazionali come prima, e quelli transazionali solo dopo il commit.
Se hai impostato tutte le configurazioni precedentemente elencate, hai configurato la consegna exactly once. Congratulazioni!
Ma c'è un ulteriore aspetto. Il transactional.id, che abbiamo configurato sopra, è in realtà il prefisso della transazione. Al gestore delle transazioni viene aggiunto un numero di sequenza. L'identificatore risultante viene emesso su transactional.id.expiration.ms, che è configurato su un cluster Kafka e ha un valore predefinito di «7 giorni». Se durante questo periodo l'applicazione non ha ricevuto alcun messaggio, allora tentare la successiva invio transazionale restituirà InvalidPidMappingException. Dopo di che, il coordinatore delle transazioni emetterà un nuovo numero di sequenza per la successiva transazione. In questo caso, il messaggio potrebbe andare perso se InvalidPidMappingException non viene gestito correttamente.
Invece dei risultati
Come puoi notare, non è sufficiente inviare messaggi a Kafka. È necessario scegliere una combinazione di parametri ed essere pronti a effettuare rapide modifiche. In questo articolo ho cercato di mostrare nel dettaglio la configurazione di exactly once delivery e ho descritto alcuni problemi di configurazione legati a client.id e transactional.id che abbiamo affrontato. Di seguito sono riportate in forma concisa le configurazioni del Producer e del Consumer.
Producer:
- acks = all
- retries > 0
- enable.idempotence = true
- max.in.flight.requests.per.connection ≤ 5 (1 — per l'invio ordinato)
- transactional.id = ${application-name}-${hostname}
Consumer:
- isolation.level = read_committed
Per minimizzare gli errori nelle applicazioni future, abbiamo creato un nostro wrapper attorno alla configurazione di spring, dove sono già impostati i valori per alcuni dei parametri elencati.
Ecco un paio di materiali per lo studio autonomo:
Fonte: habr.com
