In precedente articolo abbiamo esaminato la clusterizzazione di RabbitMQ per garantire disponibilità e tolleranza agli errori. Ora andiamo più a fondo in Apache Kafka.

Qui l'unità di replica è la partizione (partition). Ogni topic ha una o più partizioni. In ogni partizione c'è un leader con dei follower o senza di essi. Quando si crea un topic, si specifica il numero di partizioni e il fattore di replica. Un valore comune è 3, il che significa tre repliche: un leader e due follower.

Fig. 1. Quattro partizioni sono distribuite tra tre broker

Tutte le richieste di lettura e scrittura vengono inviate al leader. I follower inviano periodicamente richieste al leader per ottenere gli ultimi messaggi. I consumatori non si rivolgono mai ai follower, che esistono solo per ridondanza e tolleranza agli errori.

Guasto della partizione

Quando un broker va offline, spesso si interrompono i leader di diverse partizioni. In ognuna di esse, il leader diventa un follower di un altro nodo. In realtà, non è sempre così, poiché influisce anche il fattore di sincronizzazione: ci sono follower sincronizzati e, se non ci sono, è permesso il passaggio a una replica non sincronizzata. Ma per ora non complicchiamo le cose.

Il broker 3 esce dalla rete e per la partizione 2 viene eletto un nuovo leader sul broker 2.

Fig. 2. Il broker 3 muore e il suo follower sul broker 2 viene eletto nuovo leader della partizione 2.

Poi il broker 1 esce e anche la partizione 1 perde il suo leader, il cui ruolo passa al broker 2.

Fig. 3. Rimane solo un broker. Tutti i leader si trovano su un unico broker con zero ridondanza.

Quando il broker 1 torna in rete, aggiunge quattro follower, garantendo una certa ridondanza a ciascuna partizione. Ma tutti i leader sono ancora rimasti sul broker 2.

Fig. 4. I leader rimangono sul broker 2.

Quando il broker 3 si riattiva, torniamo a tre repliche per partizione. Ma tutti i leader sono ancora sul broker 2.

Fig. 5. Distribuzione sbilanciata dei leader dopo il ripristino dei broker 1 e 3.

Kafka offre uno strumento per un bilanciamento dei leader di qualità superiore rispetto a RabbitMQ. In quest'ultimo era necessario utilizzare un plugin di terze parti o uno script per modificare le politiche di migrazione del nodo principale, riducendo la ridondanza durante la migrazione. Inoltre, per code di grandi dimensioni era necessario accettare l'inaccessibilità durante la sincronizzazione.

Kafka introduce il concetto di "repliche preferenziali" per il ruolo di leader. Quando vengono creati i topic, Kafka cerca di distribuire i leader uniformemente tra i nodi e segna questi primi leader come preferenziali. Nel tempo, a causa di riavvii dei server, guasti e problemi di connettività, i leader possono trovarsi su altri nodi, come nel caso estremo descritto sopra.

Per risolvere questo problema, Kafka offre due opzioni:

• Opzione auto.leader.rebalance.enable=true consente al nodo controller di ripristinare automaticamente i leader alle repliche preferenziali, ripristinando così una distribuzione uniforme.
• L'amministratore può eseguire lo script kafka-preferred-replica-election.sh per rieseguire manualmente la nomina.

Fig. 6. Repliche dopo il bilanciamento

Questa era una versione semplificata del guasto, ma la realtà è più complessa, anche se non c'è nulla di troppo difficile qui. Tutto si riduce a repliche sincronizzate (In-Sync Replicas, ISR).

Repliche sincronizzate (ISR)

L'ISR è un insieme di repliche di una partizione che è considerato "sincronizzato" (in-sync). Qui c'è un leader, e potrebbero non esserci follower. Un follower è considerato sincronizzato se ha fatto copie esatte di tutti i messaggi del leader entro l'intervallo di tempo stabilito. replica.lag.time.max.ms.

Un follower viene rimosso dall'insieme ISR se:

• non ha effettuato una richiesta di fetch entro l'intervallo replica.lag.time.max.ms (considerato morto)
• non è riuscito ad aggiornarsi entro l'intervallo replica.lag.time.max.ms (considerato lento)

I follower effettuano richieste di fetch nell'intervallo replica.fetch.wait.max.ms, che per impostazione predefinita è di 500 ms.

Per spiegare chiaramente l'obiettivo dell'ISR, è necessario esaminare le conferme dal produttore (producer) e alcuni scenari di guasto. I produttori possono scegliere quando il broker invia una conferma:

• acks=0, la conferma non viene inviata
• acks=1, la conferma viene inviata dopo che il leader ha registrato il messaggio nel proprio log locale
• acks=all, la conferma viene inviata dopo che tutte le repliche nell'ISR hanno registrato il messaggio nei log locali

Nella terminologia di Kafka, se l'ISR ha memorizzato il messaggio, avviene il suo "commit". Acks=all è l'opzione più sicura, ma comporta anche una latenza aggiuntiva. Esaminiamo due esempi di errore e come le diverse opzioni di ‘acks’ interagiscono con il concetto di ISR.

Acks=1 e ISR

In questo esempio vedremo che se il leader non attende che ogni messaggio venga memorizzato da tutti i follower, è possibile che si verifichi una perdita di dati in caso di errore del leader. Il passaggio a un follower non sincronizzato può essere consentito o vietato tramite l'impostazione unclean.leader.election.enable.

In questo esempio, il produttore ha impostato acks=1. La partizione è distribuita su tutti e tre i broker. Il broker 3 è in ritardo, si è sincronizzato con il leader otto secondi fa e ora è in ritardo di 7456 messaggi. Il broker 1 è in ritardo di solo un secondo. Il nostro produttore invia un messaggio e riceve rapidamente un ack, senza sovraccarico per i follower lenti o morti, che il leader non attende.

Fig. 7. ISR con tre repliche

Il broker 2 va in errore, e il produttore riceve un errore di connessione. Dopo il passaggio della leadership al broker 1, perdiamo 123 messaggi. Il follower sul broker 1 era nell'ISR, ma non si era completamente sincronizzato con il leader quando è caduto.

Fig. 8. I messaggi vengono persi in caso di guasto

Nella configurazione bootstrap.servers al produttore sono elencati diversi broker, e può chiedere a un altro broker chi è diventato il nuovo leader della partizione. Poi stabilisce una connessione con il broker 1 e continua a inviare messaggi.

Fig. 9. L'invio dei messaggi riprende dopo una breve interruzione

Il broker 3 è in ulteriore ritardo. Effettua richieste di recupero, ma non riesce a sincronizzarsi. Questo può essere dovuto a una connessione di rete lenta tra i broker, problemi di archiviazione, ecc. Viene rimosso dall'ISR. Ora l'ISR consiste in una sola replica: il leader! Il produttore continua a inviare messaggi e a ricevere conferme.

Fig. 10. Il follower sul broker 3 viene rimosso dall'ISR

Il broker 1 va in giù e il ruolo di leader passa al broker 3 con una perdita di 15286 messaggi! Il produttore riceve un errore di connessione. Il passaggio al leader al di fuori dell'ISR è stato possibile solo grazie alla configurazione. unclean.leader.election.enable=true. Se impostato su false, il passaggio non sarebbe avvenuto e tutte le richieste di lettura e scrittura sarebbero state rifiutate. In tal caso, aspettiamo il ritorno del broker 1 con i suoi dati intatti nella replica, che riprenderà la leadership.

Fig. 11. Il broker 1 va in giù. Durante il fallimento si perdono un gran numero di messaggi.

Il produttore stabilisce una connessione con l'ultimo broker e vede che ora è il leader della partizione. Inizia a inviare messaggi al broker 3.

Fig. 12. Dopo una breve interruzione, i messaggi ricominciano a essere inviati alla partizione 0.

Abbiamo visto che, oltre a brevi interruzioni per stabilire nuove connessioni e cercare un nuovo leader, il produttore ha continuato a inviare messaggi. Questa configurazione garantisce disponibilità a scapito della coerenza (sicurezza dei dati). Kafka ha perso migliaia di messaggi, ma ha continuato a ricevere nuove registrazioni.

Acks=all e ISR

Ripetiamo questo scenario ancora una volta, ma con acks=all. Il broker 3 ha in media un ritardo di quattro secondi. Il produttore invia un messaggio con acks=all, e ora non riceve una risposta veloce. Il leader aspetta che il messaggio venga salvato da tutte le repliche in ISR.

Fig. 13. ISR con tre repliche. Una funziona lentamente, il che porta a un ritardo nella registrazione

Dopo quattro secondi di ulteriore ritardo, il broker 2 invia un ack. Tutte le repliche sono ora completamente aggiornate.

Fig. 14. Tutte le repliche memorizzano i messaggi e viene inviato ack

Il broker 3 ora è ulteriormente in ritardo ed è rimosso dall'ISR. Il ritardo diminuisce notevolmente, poiché non ci sono più repliche lente nell'ISR. Il broker 2 sta ora aspettando solo il broker 1, che ha un ritardo medio di 500 ms.

Fig. 15. Replica sul broker 3 rimossa dall'ISR

Poi cade il broker 2, e la leadership passa al broker 1 senza perdita di messaggi.

Fig. 16. Broker 2 in caduta

Il produttore trova un nuovo leader e inizia a inviargli messaggi. Il ritardo diminuisce ulteriormente, dato che ora l'ISR è composto da una sola replica! Pertanto, l'opzione acks=all non aggiunge ridondanza.

Fig. 17. Replica sul broker 1 assume la leadership senza perdita di messaggi

Poi cade il broker 1, e la leadership passa al broker 3 con una perdita di 14238 messaggi!

Fig. 18. Il broker 1 muore, e il passaggio di leadership con la configurazione unclean porta a una significativa perdita di dati

Potremmo non impostare l'opzione unclean.leader.election.enable su true. Per impostazione predefinita è false. La configurazione acks=all con unclean.leader.election.enable=true garantisce la disponibilità con una certa sicurezza aggiuntiva dei dati. Ma, come puoi vedere, possiamo comunque perdere i messaggi.

Ma cosa succede se vogliamo aumentare la sicurezza dei dati? Possiamo impostare unclean.leader.election.enable = false, ma questo non ci proteggerà necessariamente dalla perdita di dati. Se il leader è andato giù in modo critico e ha portato via i dati, i messaggi sono comunque persi, e inoltre viene persa la disponibilità finché l'amministratore non ripristina la situazione.

È meglio garantire la ridondanza di tutti i messaggi, altrimenti rinunciare alla registrazione. In questo modo, almeno dal punto di vista del broker, la perdita di dati è possibile solo in caso di due o più guasti simultanei.

Acks=all, min.insync.replicas e ISR

Con la configurazione del topic min.insync.replicas aumentiamo il livello di sicurezza dei dati. Rivediamo l'ultima parte dello scenario precedente, ma questa volta con min.insync.replicas=2.

Quindi, il broker 2 ha un leader replica, e il follower sul broker 3 è stato rimosso dall'ISR.

Fig. 19. ISR di due repliche

Il broker 2 si interrompe, mentre la leadership passa al broker 1 senza perdita di messaggi. Ma ora l'ISR è composto da una sola replica. Ciò non soddisfa il numero minimo necessario per registrare i dati, quindi il broker risponde con un errore alla richiesta di scrittura. NotEnoughReplicas.

Fig. 20. Il numero di ISR è uno in meno rispetto a quanto indicato in min.insync.replicas.

Questa configurazione sacrifica la disponibilità per la coerenza. Prima di confermare un messaggio, garantiamo che venga registrato su almeno due repliche. Questo offre al produttore una maggiore sicurezza. Qui la perdita di messaggi è possibile solo in caso di guasto simultaneo di due repliche in un breve lasso di tempo, prima che il messaggio venga replicato a un follower aggiuntivo, il che è poco probabile. Ma se sei molto paranoico, puoi impostare un fattore di replica di 5. min.insync.replicas e 3. Qui tre broker devono fallire simultaneamente per perdere la registrazione! Certamente, per questa affidabilità pagherai un ritardo aggiuntivo.

Quando la disponibilità è necessaria per la sicurezza dei dati

Come in nel caso di RabbitMQ, a volte la disponibilità è necessaria per la sicurezza dei dati. Devi considerare quanto segue:

• Un publisher può semplicemente restituire un errore e il servizio superiore o l'utente può riprovare più tardi?
• Può un publisher memorizzare un messaggio localmente o in un database per riprovare in seguito?

Se la risposta è negativa, l'ottimizzazione della disponibilità aumenta la sicurezza dei dati. Si perderanno meno dati scegliendo la disponibilità invece del rifiuto della registrazione. In questo modo, tutto si riduce alla ricerca di un equilibrio, e la decisione dipende dalla situazione specifica.

Significato di ISR

Il set di ISR consente di trovare il giusto equilibrio tra sicurezza dei dati e latenza. Ad esempio, garantire la disponibilità in caso di guasto della maggior parte delle repliche, minimizzando l'impatto delle repliche morte o lente in termini di latenza.

Scegliamo noi stessi il valore replica.lag.time.max.ms in base alle nostre esigenze. In sostanza, questo parametro indica quale latenza siamo disposti ad accettare. acks=allIl valore predefinito è di dieci secondi. Se per te è troppo lungo, puoi ridurlo. In tal caso, aumenterà la frequenza delle modifiche in ISR, poiché i follower verranno rimossi e aggiunti più frequentemente.

In RabbitMQ ci sono solo insiemi di specchi da replicare. Gli specchi lenti introducono un ulteriore ritardo, e la risposta dei mirror morti può essere attesa fino alla scadenza del tempo di vita dei pacchetti che verificano la disponibilità di ciascun nodo (net tick). ISR è un modo interessante per evitare questi problemi di aumento del ritardo. Ma rischiamo di perdere ridondanza, poiché ISR può ridursi solo al leader. Per evitare questo rischio, utilizzare la configurazione min.insync.replicas.

Garanzia di connessione per i clienti

Nelle impostazioni bootstrap.servers del produttore e del consumatore è possibile specificare più broker per la connessione dei clienti. L'idea è che, in caso di disconnessione di un nodo, rimangano diversi di riserva con cui il cliente può stabilire una connessione. Questi non devono necessariamente essere i leader delle partizioni, ma semplicemente un punto di partenza per il caricamento iniziale. Il cliente può chiedere loro su quale nodo è situato il leader della partizione per lettura/scrittura.

In RabbitMQ, clients can connect to any node, and the internal routing sends requests where they need to go. This means you can place a load balancer in front of RabbitMQ. Kafka requires clients to connect to the node where the leader of the corresponding partition is hosted. In this situation, a load balancer cannot be placed. The list bootstrap.servers is critical for clients to access the necessary nodes and locate them after a failure.

Kafka Consensus Architecture

So far, we have not discussed how a cluster becomes aware of a broker failure and how a new leader is selected. To understand how Kafka operates with network partitions, it is essential first to grasp the consensus architecture.

Each Kafka cluster is deployed along with a Zookeeper cluster — a distributed consensus service that allows the system to achieve consensus on some defined state, prioritizing consistency over availability. Approval from the majority of Zookeeper nodes is required for read and write operations.

Zookeeper stores the state of the cluster:

• Elenco dei topic, delle partizioni, della configurazione, delle repliche del leader attuale e delle repliche preferite.
• Membri del cluster. Ogni broker invia un ping al cluster Zookeeper. Se quest'ultimo non riceve un ping per un periodo di tempo definito, registra il broker come non disponibile.
• Selezione dei nodi primari e secondari per il controller.

Il nodo controller è uno dei broker Kafka responsabile dell'elezione dei leader delle repliche. Zookeeper invia notifiche al controller riguardo ai cambiamenti nella composizione del cluster e nei topic, e il controller deve agire in base a queste modifiche.

Ad esempio, consideriamo un nuovo topic con dieci partizioni e un fattore di replica di 3. Il controller deve scegliere un leader per ciascuna partizione, cercando di distribuire ottimamente i leader tra i broker.

Per ciascuna partizione, il controller:

• aggiorna le informazioni in Zookeeper riguardo a ISR e leader;
• invia il comando LeaderAndISRCommand a ogni broker che ospita una replica di quella partizione, informando i broker su ISR e leader.

Quando un broker crolla, Zookeeper invia una notifica al controller, il quale sceglie un nuovo leader. Anche in questo caso, il controller aggiorna prima Zookeeper e poi invia un comando a ciascun broker, informandoli del cambiamento di leadership.

Ogni leader è responsabile dell'insieme di ISR. La configurazione replica.lag.time.max.ms definisce chi ne farà parte. Quando l'ISR cambia, il leader comunica a Zookeeper le nuove informazioni.

Zookeeper è sempre informato su qualsiasi cambiamento, in modo che in caso di guasti la leadership possa passare senza problemi a un nuovo leader.

Fig. 21. Consenso Kafka

Protocollo di replica

Comprendere i dettagli della replica aiuta a capire meglio i potenziali scenari di perdita di dati.

Richieste di fetch, Log End Offset (LEO) e Highwater Mark (HW)

Abbiamo visto che i follower inviano periodicamente richieste di fetch al leader. L'intervallo predefinito è di 500 ms. Questo è diverso da RabbitMQ, dove la replica è avviata non dal mirror della coda, ma dal master. Il master invia le modifiche ai mirror.

Il leader e tutti i follower mantengono il Log End Offset (LEO) e il segno Highwater (HW). Il segno LEO memorizza l'offset dell'ultimo messaggio nella replica locale, mentre HW rappresenta l'offset dell'ultimo commit. Ricordate che per lo stato 'commit' il messaggio deve essere salvato in tutte le repliche ISR. Questo significa che di solito LEO precede HW.

Quando il leader riceve un messaggio, lo memorizza localmente. Un follower fa una richiesta di recupero, passando il proprio LEO. Successivamente, il leader invia un pacchetto di messaggi, a partire da questo LEO, e comunica l'attuale HW. Quando il leader riceve informazioni che tutte le repliche hanno salvato il messaggio con l'offset specificato, sposta il segno HW. Solo il leader può spostare HW, quindi tutti i follower conoscono il valore attuale nelle risposte alla loro richiesta. Questo significa che i follower possono essere in ritardo rispetto al leader sia per i messaggi che per la conoscenza di HW. I consumatori ricevono messaggi solo fino all'attuale HW.

Si prega di notare che "persistito" significa memorizzato in memoria, non su disco. Per motivi di prestazioni, Kafka sincronizza su disco a intervalli prestabiliti. Anche RabbitMQ ha un intervallo simile, ma invierà la conferma al publisher solo dopo che il master e tutte le repliche hanno memorizzato il messaggio su disco. Gli sviluppatori di Kafka, per motivi di prestazioni, hanno deciso di inviare l'ack non appena il messaggio è memorizzato in memoria. Kafka si affida al fatto che la ridondanza compensi il rischio di mantenere i messaggi confermati solo in memoria per un breve periodo.

Guasto del leader

Quando il leader cade, Zookeeper avvisa il controller, e questo sceglie una nuova replica come leader. Il nuovo leader stabilisce un nuovo watermark HW in base al suo LEO. Le informazioni sul nuovo leader vengono quindi ricevute dai follower. A seconda della versione di Kafka, il follower sceglierà uno dei due scenari:

1. Troncherà il log locale fino a un HW noto e invierà una richiesta al nuovo leader per i messaggi dopo quella marcatura.
2. Invierà una richiesta al leader per conoscere l'HW al momento della sua elezione, quindi trimmerà il log fino a quel punto. Successivamente inizierà a fare richieste periodiche di campionamento, partendo da questa posizione.

Un follower potrebbe dover trimmare il log per i seguenti motivi:

• Quando si verifica un fallimento del leader, il primo follower del set ISR, registrato in Zookeeper, vince le elezioni e diventa leader. Tutti i follower nell'ISR, sebbene considerati "sincronizzati", potrebbero non aver ricevuto dal precedente leader copie di tutti i messaggi. È possibile che il follower eletto non abbia la copia più aggiornata. Kafka garantisce che non ci siano discrepanze tra le repliche. Pertanto, per evitare discrepanze, ogni follower deve trimmare il proprio log fino al valore HW del nuovo leader al momento della sua elezione. Questa è un'altra ragione per cui la configurazione acks=all è così importante per la coerenza.
• I messaggi vengono registrati periodicamente su disco. Se tutti i nodi del cluster falliscono contemporaneamente, le repliche con offset diversi verranno salvate sui dischi. È possibile che, quando i broker torneranno in rete, un nuovo leader, che sarà eletto, si trovi indietro rispetto ai suoi follower, poiché è stato salvato su disco prima degli altri.

Riconnessione al cluster

Durante la riconnessione al cluster, le repliche si comportano come in caso di fallimento del leader: controllano la replica del leader e troncheranno il loro log fino al suo HW (al momento dell'elezione). A differenza di ciò, RabbitMQ considera i nodi riconnessi come completamente nuovi. In entrambi i casi, il broker scarta qualsiasi stato esistente. Se viene utilizzata la sincronizzazione automatica, il master deve replicare assolutamente tutto il contenuto attuale in un nuovo mirror con il metodo 'e che il mondo aspetti'. Durante questa operazione, il master non accetta alcuna operazione di lettura o scrittura. Questo approccio crea problemi in grandi code.

Kafka è un log distribuito e, in generale, conserva più messaggi rispetto alla coda RabbitMQ, dove i dati vengono rimossi dalla coda dopo essere stati letti. Le code attive devono rimanere relativamente piccole. Ma Kafka è un log con una propria politica di conservazione, che può stabilire un termine di giorni o settimane. L'approccio con blocco della coda e sincronizzazione totale è completamente inaccettabile per un log distribuito. Invece, i follower di Kafka semplicemente accorciano il loro log fino al leader HW (al momento della sua elezione) se la loro copia supera il leader. Nel caso più probabile, quando un follower è in ritardo, inizia semplicemente a fare richieste di polling, partendo dal suo attuale LEO.

I nuovi follower o quelli riuniti iniziano al di fuori dell'ISR e non partecipano ai commit. Lavorano semplicemente accanto al gruppo, ricevendo messaggi il più rapidamente possibile, finché non raggiungono il leader e non entrano nell'ISR. Qui non ci sono blocchi e non è necessario scartare tutti i propri dati.

Violazione della connettività

Kafka ha più componenti rispetto a RabbitMQ, quindi si presenta un insieme più complesso di comportamenti quando si verifica una perdita di connettività nel cluster. Tuttavia, Kafka è stato progettato fin dall'inizio per i cluster, quindi le soluzioni sono ben ragionate.

Di seguito sono riportati alcuni scenari di perdita di connettività:

• Scenario 1. Il follower non vede il leader, ma vede ancora Zookeeper.
• Scenario 2. Il leader non vede alcun follower, ma vede ancora Zookeeper.
• Scenario 3. Il follower vede il leader, ma non vede Zookeeper.
• Scenario 4. Il leader vede i follower, ma non vede Zookeeper.
• Scenario 5. Il follower è completamente isolato sia dagli altri nodi Kafka sia da Zookeeper.
• Scenario 6. Il leader è completamente isolato sia dagli altri nodi Kafka sia da Zookeeper.
• Scenario 7. Il nodo controller di Kafka non vede un altro nodo Kafka.
• Scenario 8. Il controller di Kafka non vede Zookeeper.

Per ogni scenario è previsto un comportamento specifico.

Scenario 1. Il follower non vede il leader, ma vede ancora Zookeeper

Fig. 22. Scenario 1. ISR di tre repliche

La perdita di connettività separa il broker 3 dai broker 1 e 2, ma non da Zookeeper. Il broker 3 non può più inviare richieste di fetch. Dopo un certo periodo di tempo replica.lag.time.max.ms viene rimosso dall'ISR e non partecipa ai commit dei messaggi. Non appena la connettività viene ripristinata, riprenderà le richieste di fetch e si unirà all'ISR quando raggiungerà il leader. Zookeeper continuerà a ricevere ping e considererà il broker vivo e vegeto.

Fig. 23. Scenario 1. Il broker viene rimosso dall'ISR se non riceve una richiesta di fetch entro l'intervallo replica.lag.time.max.ms

Non c'è alcuna separazione logica (split-brain) o pausa del nodo, come in RabbitMQ. Piuttosto, la ridondanza viene ridotta.

Scenario 2. Il leader non vede alcun follower, ma vede ancora Zookeeper

Fig. 24. Scenario 2. Leader e due follower

La violazione della connettività di rete separa il leader dai follower, ma il broker vede ancora Zookeeper. Come nel primo scenario, l'ISR si riduce, ma questa volta solo al leader, poiché tutti i follower smettono di inviare richieste di fetch. Ancora una volta, non c'è alcuna separazione logica. Invece, si verifica una perdita di ridondanza per i nuovi messaggi finché la connettività non viene ripristinata. Zookeeper continua a ricevere ping e considera il broker vivo e vegeto.

Fig. 25. Scenario 2. ISR ridotto solo al leader

Scenario 3. Il follower vede il leader, ma non vede Zookeeper

Il follower si separa da Zookeeper, ma non dal broker con il leader. Di conseguenza, il follower continua a fare richieste di polling ed è membro dell'ISR. Zookeeper non riceve più ping e registra il crash del broker, ma poiché si tratta solo di un follower, non ci sono conseguenze una volta che il broker viene ripristinato.

Fig. 26. Scenario 3. Il follower continua a inviare richieste di polling al leader

Scenario 4. Il leader vede i follower, ma non vede Zookeeper

Fig. 27. Scenario 4. Il leader e due follower

Il leader è separato da Zookeeper, ma non dai broker con i follower.

Fig. 28. Scenario 4. Il leader è isolato da Zookeeper

Dopo un po' Zookeeper registrerà il crash del broker e ne informerà il controller. Questi sceglierà un nuovo leader tra i follower. Tuttavia, il leader originale continuerà a pensare di essere il leader e continuerà a ricevere registrazioni con acks=1. I follower non gli inviano più richieste di polling, quindi lui li considererà morti e tenterà di comprimere l'ISR fino a se stesso. Ma poiché non ha un collegamento a Zookeeper, non sarà in grado di farlo e in quel momento smetterà di accettare ulteriori registrazioni.

Messaggi acks=all non riceveranno conferma, poiché l'ISR include prima tutte le repliche e i messaggi non arrivano a esse. Quando il leader originale cercherà di rimuoverli dall'ISR, non potrà farlo e smetterà completamente di ricevere messaggi.

I clienti noteranno presto il cambio di leader e inizieranno a inviare registrazioni al nuovo server. Una volta che la rete si ristabilisce, il leader originale vede che non è più il leader e riduce il proprio log al valore HW che aveva il nuovo leader al momento del guasto, per evitare incongruenze. Poi inizierà a inviare richieste di recupero al nuovo leader. Tutte le registrazioni del leader originale non replicate al nuovo leader andranno perse. Ovvero, andranno persi i messaggi non confermati dal leader originale nei pochi secondi in cui sono stati attivi due leader.

Fig. 29. Scenario 4. Il leader sul broker 1 diventa follower dopo il ripristino della rete

Scenario 5. Il follower è completamente isolato sia dagli altri nodi Kafka che da Zookeeper

Il follower è completamente isolato sia dagli altri nodi Kafka che da Zookeeper. Viene semplicemente rimosso dall'ISR fino a quando la rete non si ristabilisce, e poi recupera gli altri.

Fig. 30. Scenario 5. The isolated follower is removed from the ISR

Scenario 6. The leader is completely separated from both other Kafka nodes and Zookeeper

Fig. 31. Scenario 6. The leader and two followers

The leader is completely isolated from its followers, the controller, and Zookeeper. For a short period, it will continue to accept writes from acks=1.

Fig. 32. Scenario 6. Leader isolation from other Kafka nodes and Zookeeper

After not receiving requests for replica.lag.time.max.ms, it will attempt to shrink the ISR to itself, but it will be unable to do so due to loss of connection with Zookeeper, at which point it will stop accepting writes.

Meanwhile, Zookeeper marks the isolated broker as dead, and the controller will elect a new leader.

Fig. 33. Scenario 6. Two leaders

The original leader may accept writes for a few seconds, but will then stop accepting any messages. Clients are updated every 60 seconds with the latest metadata. They will be notified of the leadership change and will start sending writes to the new leader.

Fig. 34. Scenario 6. Producers switch to the new leader

Tutte le registrazioni confermate fatte dal leader originale dal momento della perdita di connettività andranno perse. Una volta ripristinata la rete, il leader originale attraverso Zookeeper scoprirà di non essere più il leader. Dopodiché, troncherà il proprio log fino all'HW del nuovo leader al momento della sua elezione e inizierà a inviare richieste come follower.

Fig. 35. Scenario 6. Il leader originale diventa un follower dopo il ripristino della connettività di rete

In questa situazione, per un breve periodo, può verificarsi una divisione logica, ma solo se acks=1 e min.insync.replicas anche 1. La divisione logica si conclude automaticamente o dopo il ripristino della rete, quando il leader originale comprende che non è più il leader, oppure quando tutti i client capiscono che il leader è cambiato e iniziano a scrivere al nuovo leader — a seconda di cosa accade prima. In ogni caso, ci sarà una perdita di alcuni messaggi, ma solo con acks=1.

C'è un'altra opzione per questo scenario, in cui prima della divisione della rete i follower rimangono indietro e il leader riduce l'ISR a se stesso. Poi si isola a causa della perdita di connettività. Viene eletto un nuovo leader, ma il leader iniziale continua a ricevere registrazioni, anche acks=all, perché nell'ISR non c'è nessun altro. Queste registrazioni andranno perse dopo il ripristino della rete. L'unico modo per evitare questo scenario è min.insync.replicas = 2.

Scenario 7. Il nodo controller di Kafka non vede un altro nodo Kafka

In generale, dopo la perdita di connessione con un nodo Kafka, il controller non sarà in grado di trasmettergli alcuna informazione riguardo alla modifica del leader. Nella peggiore delle ipotesi, ciò porterà a una breve divisione logica, come nello scenario 6. Più frequentemente, il broker non diventerà semplicemente candidato alla leadership in caso di guasto dell'ultimo.

Scenario 8. Il controller di Kafka non vede Zookeeper

Un controller Zookeeper disconnesso non riceverà ping e selezionerà un nuovo nodo Kafka come controller. Il controller originale può continuare a presentarsi come tale, ma non riceve notifiche da Zookeeper, quindi non avrà né compiti né responsabilità. Una volta che la rete sarà ripristinata, capirà di non essere più un controller, ma un normale nodo Kafka.

Conclusioni dagli scenari

Osserviamo che la perdita di connettività dei follower non porta alla perdita di messaggi, ma riduce temporaneamente la ridondanza fino a quando la rete non si ripristina. Questo, ovviamente, può portare a una perdita di dati se uno o più nodi risultano persi.

Se, a causa della perdita di connettività, il leader viene separato da Zookeeper, ciò può portare a una perdita di messaggi con acks=1. L'assenza di connessione con Zookeeper provoca una breve divisione logica con due leader. Questo problema viene risolto dal parametro acks=all.

Caratteristica min.insync.replicas in due o più repliche fornisce garanzie aggiuntive che tali scenari a breve termine non porteranno alla perdita di messaggi, come nello scenario 6.

Riepilogo sulla perdita di messaggi

Elenchiamo tutti i modi in cui è possibile perdere dati in Kafka:

• Qualsiasi guasto del leader, se i messaggi venivano confermati tramite acks=1
• Qualsiasi passaggio di leadership sporco (unclean), ovvero a un follower al di fuori dell'ISR, anche con acks=all
• Isolamento del leader da Zookeeper, se i messaggi venivano confermati tramite acks=1
• Isolamento totale del leader, che ha già ridotto il gruppo ISR a se stesso. Tutti i messaggi andranno persi, anche acks=all. Questo è vero solo se min.insync.replicas=1.
• Guasti simultanei di tutti i nodi della partizione. Poiché i messaggi vengono confermati dalla memoria, alcuni potrebbero non essere ancora scritti su disco. Dopo il riavvio dei server, potrebbero mancare alcuni messaggi.

I passaggi sporchi di leadership possono essere evitati, vietandoli oppure garantendo una ridondanza di almeno due. La configurazione più robusta è una combinazione di acks=all e min.insync.replicas più di 1.

Confronto diretto dell'affidabilità di RabbitMQ e Kafka

Per garantire affidabilità e alta disponibilità, entrambe le piattaforme implementano un sistema di replica primaria e secondaria. Tuttavia, RabbitMQ ha un tallone d'Achille. Durante la riconnessione dopo un guasto, i nodi scartano i loro dati e la sincronizzazione si blocca. Questo doppio colpo mette in dubbio la longevità delle grandi code in RabbitMQ. Dovrete accettare o una riduzione della ridondanza o lunghe fasi di blocco. La diminuzione della ridondanza aumenterà il rischio di una massiccia perdita di dati. Tuttavia, se le code sono piccole, per garantire la ridondanza con brevi periodi di inattività (alcuni secondi), si può affrontare attraverso tentativi di riconnessione.

In Kafka, this problem doesn't exist. It only drops data from the point of divergence between the leader and the follower. All shared data is preserved. Furthermore, replication does not block the system. The leader continues to accept records while the new follower catches up, making joining or reconnecting to the cluster a trivial task for DevOps. Of course, issues like network bandwidth when replicating still remain. If multiple followers are added simultaneously, one may encounter bandwidth limits.

RabbitMQ surpasses Kafka in reliability during simultaneous failures of multiple servers in the cluster. As we mentioned, RabbitMQ sends a confirmation to the publisher only after the message is written to disk by the master and all mirrors. However, this introduces additional latency for two reasons:

• fsync every few hundred milliseconds
• A mirror failure can only be detected after the expiration of the packet lifespan that checks the availability of each node (net tick). If a mirror lags or fails, it adds to the delay.

Kafka scommette sul fatto che se un messaggio è memorizzato su più nodi, è possibile confermare i messaggi non appena entrano in memoria. Questo comporta un rischio di perdita di qualsiasi tipo di messaggio (anche acks=all, min.insync.replicas=2) in caso di guasto simultaneo.

In generale, Kafka dimostra prestazioni superiori ed è stato progettato fin dall'inizio per i cluster. Il numero di follower può essere aumentato fino a 11, se necessario per l'affidabilità. Un fattore di replica di 5 e un numero minimo di repliche in stato sincronizzato min.insync.replicas=3 renderanno la perdita di messaggi un evento molto raro. Se la tua infrastruttura è in grado di garantire tale fattore di replica e livello di ridondanza, puoi optare per questa opzione.

La clusterizzazione di RabbitMQ è utile per piccole code. Ma anche piccole code possono crescere rapidamente con alto traffico. Quando le code diventano grandi, sarà necessario fare una scelta difficile tra disponibilità e affidabilità. La clusterizzazione di RabbitMQ è più adatta a situazioni non tipiche, dove i vantaggi di flessibilità di RabbitMQ superano eventuali svantaggi della sua clusterizzazione.

Una delle soluzioni per la vulnerabilità di RabbitMQ riguardo le code grandi è suddividerle in più code più piccole. Se non è necessario un ordinamento completo dell'intera coda, ma solo dei messaggi rilevanti (ad esempio, messaggi di un cliente specifico), oppure non ordinare affatto, questa opzione è accettabile: dai un'occhiata al mio progetto Rebilanciatore per suddividere la coda (il progetto è ancora in fase iniziale).

Infine, non dimenticate una serie di bug nei meccanismi di clustering e replica sia di RabbitMQ che di Kafka. Col tempo, i sistemi sono diventati più maturi e stabili, ma nessun messaggio sarà mai al 100% protetto dalla perdita! Inoltre, nei data center possono verificarsi disastri su larga scala!

Se ho tralasciato qualcosa, ho commesso un errore o non sei d'accordo con uno qualsiasi dei punti, non esitare a lasciare un commento o contattarmi.

Spesso mi chiedono: «Cosa scegliere, Kafka o RabbitMQ?», «Qual è la piattaforma migliore?». La verità è che dipende davvero dalla tua situazione, esperienza attuale, e così via. Non mi sento di esprimere un'opinione, poiché sarebbe troppo riduttivo raccomandare una sola piattaforma per tutte le possibili applicazioni e limitazioni. Ho scritto questo ciclo di articoli affinché possiate formarvi un'opinione personale.

Voglio dire che entrambe le soluzioni sono leader nel loro campo. Forse sono un po' di parte, poiché per esperienza personale nei miei progetti, tendo a dare valore a caratteristiche come l'ordinamento garantito dei messaggi e l'affidabilità.

Osservo altre tecnologie che mancano di questa affidabilità e ordinamento garantito, poi guardo a RabbitMQ e Kafka — e comprendo l'incredibile valore di entrambi questi sistemi.

Fonte: habr.com