Ciao, Habr!
Lavoro nel team di Tinkoff che si occupa dello sviluppo del nostro centro notifiche. Per la maggior parte sviluppo in Java utilizzando Spring Boot e affronto varie problematiche tecniche che sorgono nel progetto.
La maggior parte dei nostri microservizi interagiscono tra loro in modo asincrono attraverso un broker di messaggi. In passato abbiamo utilizzato IBM MQ come broker, ma ha smesso di gestire il carico, pur offrendo elevate garanzie di consegna.
Come sostituzione ci è stata proposta Apache Kafka, che presenta un alto potenziale di scalabilità, ma purtroppo richiede un approccio praticamente personalizzato alla configurazione per diversi scenari. Inoltre, il meccanismo di consegna at least once, che opera in Kafka per impostazione predefinita, non consentiva di mantenere il livello di coerenza necessario di default. Qui condividerò la nostra esperienza con la configurazione di Kafka, in particolare parlerò di come impostare e vivere con la consegna exactly once.
Consegna garantita e non solo
I parametri di cui parleremo di seguito aiuteranno a prevenire una serie di problemi con le impostazioni di connessione predefinite. Ma prima vogliamo concentrarci su un parametro che faciliterà eventuali debug.
A questo aiuterà client.id per Producer e Consumer. A prima vista, come valore si può utilizzare il nome dell'applicazione, e nella maggior parte dei casi funzionerà. Tuttavia, una situazione in cui l'applicazione utilizza più Consumer e si assegna loro lo stesso client.id porta al seguente avviso:
org.apache.kafka.common.utils.AppInfoParser — Errore di registrazione 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 del nome dell'applicazione e, ad esempio, del nome del topic. Il risultato della nostra configurazione può essere visualizzato nell'output del comando kafka-consumer-groups dalle utility di Confluent:
Ora esaminiamo lo scenario della consegna garantita di un messaggio. Un Kafka Producer ha un parametro acks, che permette di configurare dopo quanti acknowledge il leader del cluster deve considerare il messaggio registrato con successo. Questo parametro può assumere i seguenti valori:
- 0 — gli acknowledge non verranno considerati.
- 1 — parametro di default, necessario acknowledge solo da 1 replica.
- −1 — richiesti acknowledge da tutte le repliche sincronizzate (configurazione del cluster. min.insync.replicas).
Dai valori elencati, è evidente che acks uguale a −1 offre le garanzie più forti che il messaggio non venga perso.
Come tutti sappiamo, i sistemi distribuiti non sono affidabili. Per proteggersi da guasti temporanei, Kafka Producer fornisce il parametro retries, che consente di impostare il numero di tentativi di reinvio durante delivery.timeout.ms. Poiché il parametro retries ha come valore predefinito Integer.MAX_VALUE (2147483647), il numero di reinvii del messaggio può essere regolato modificando solo delivery.timeout.ms.
Passiamo a una consegna exactly once
Le impostazioni elencate consentono al nostro Producer di inviare messaggi con un'alta garanzia. Parliamo ora 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. I requisiti preliminari per attivare 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 sopra indicati.
Quando l'idempotenza è configurata, bisogna garantire che messaggi identici finiscano sempre nelle stesse partizioni. Questo può essere fatto configurando la chiave e il parametro partitioner.class nel Producer. Iniziamo con la chiave. Per ogni invio deve essere la stessa. È facile ottenerlo utilizzando un identificatore aziendale dal messaggio originale. Il parametro partitioner.class ha un valore predefinito — . Con questa strategia di partizionamento predefinita, procediamo in questo modo:
- Se la partizione è specificata esplicitamente durante l'invio del messaggio, la utilizziamo.
- Se la partizione non è specificata ma è fornita una chiave, scegliamo la partizione in base all'hash della chiave.
- Se né la partizione né la chiave sono specificate, selezioniamo le partizioni in modo sequenziale (round-robin).
Inoltre, l'uso di una chiave e l'invio idempotente con il parametro max.in.flight.requests.per.connection = 1 assicura un'elaborazione ordinata dei messaggi nel Consumer. È importante ricordare che, se nel tuo cluster è attivata la gestione degli accessi, avrai bisogno di diritti per la registrazione idempotente nel topic.
Se ti manca la possibilità di invio idempotente per chiave o la logica lato Producer richiede di mantenere la coerenza dei dati tra diverse partizioni, le transazioni possono tornarti utili. Inoltre, tramite una transazione a catena, è possibile sincronizzare condizionatamente la scrittura in Kafka, ad esempio, con la registrazione in un DB. Per abilitare l'invio transazionale al Producer, è necessario che esso possieda idempotenza e specificare ulteriormente transactional.id. Se nel tuo cluster Kafka è abilitata la gestione degli accessi, per la registrazione delle transazioni, come per quelle idempotenti, saranno necessari diritti di scrittura, che possono essere concessi tramite una maschera utilizzando il valore memorizzato in transactional.id.
Formalmente, qualsiasi stringa può essere usata come identificatore di 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à arrestata con un errore, poiché Kafka la considererà un processo zombie.
org.apache.kafka.common.errors.ProducerFencedException: Il produttore ha tentato un'operazione con un'epoca obsoleta. O c'è un produttore più recente con lo stesso transactionalId, oppure la transazione del produttore è scaduta dal broker.Per risolvere questo problema, aggiungiamo al nome dell'applicazione un suffisso che corrisponde al nome host, ottenuto dalle variabili d'ambiente.
Il produttore è configurato, ma le transazioni su Kafka gestiscono solo la visibilità del messaggio. Indipendentemente dallo stato della transazione, il messaggio viene immediatamente inserito nel topic, ma ha attributi di sistema aggiuntivi.
Per evitare che tali messaggi vengano letti dal Consumer in anticipo, è necessario impostare il parametro isolation.level su read_committed. Questo Consumer potrà leggere i messaggi non transazionali come prima, mentre i messaggi transazionali solo dopo il commit.
Se hai configurato tutte le impostazioni precedentemente elencate, hai impostato la consegna exactly once. Congratulazioni!
Ma c'è un ulteriore dettaglio. Il transactional.id, che abbiamo configurato sopra, è in realtà il prefisso della transazione. Un numero di sequenza viene aggiunto dal gestore delle transazioni. L'identificativo risultante viene emesso su transactional.id.expiration.ms, che è configurato nel cluster Kafka e ha un valore di default di "7 giorni". Se entro questo periodo l'applicazione non ha ricevuto messaggi, la prossima volta che tenterai di inviare una transazione otterrai InvalidPidMappingException. A questo punto, il coordinatore delle transazioni assegnerà un nuovo numero di sequenza per la prossima transazione. In questo caso, il messaggio potrebbe andare perso se l'InvalidPidMappingException non viene gestito correttamente.
Invece dei risultati
Come puoi notare, non è sufficiente semplicemente inviare messaggi in Kafka. È necessario scegliere una combinazione di parametri e essere pronti a effettuare modifiche rapide. In questo articolo, ho cercato di mostrare in dettaglio come configurare la consegna exactly once e ho descritto alcuni problemi relativi alle configurazioni di client.id e transactional.id con cui abbiamo avuto a che fare. Di seguito sono riportate in breve le impostazioni 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 sulla 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
