W W poprzednim artykule Zajmiemy się klasteryzacją RabbitMQ w celu zapewnienia niezawodności i dużej dostępności. Teraz dokładniej przyjrzymy się Apache Kafka.

Jednostką replikacji jest tutaj partycja. Każdy temat ma jedną lub więcej partycji. W każdej partycji jest lider oraz followerzy lub ich brak. Podczas tworzenia tematu określa się liczbę partycji i współczynnik replikacji. Zwykle wartość wynosi 3, co oznacza trzy repliki: jeden lider i dwóch followerów.

Rys. 1. Cztery partycje rozdzielone między trzema brokerami.

Wszystkie zapytania o odczyt i zapis trafiają do lidera. Followerzy okresowo wysyłają do lidera zapytania o najnowsze wiadomości. Konsumenci nigdy nie kontaktują się z followerami, ci istnieją tylko dla nadmiarowości i niezawodności.

Awaria partycji.

Gdy broker ulega awarii, często liderzy kilku partycji przestają działać. W każdym z nich liderem zostaje follower z innego węzła. W rzeczywistości nie zawsze tak jest, ponieważ wpływa na to również czynnik synchronizacji: czy są synchronizowane followerzy, a jeśli nie, to czy przejście na niesynchronizowaną replikę jest dozwolone. Ale na razie nie komplikujmy.

Broker 3 wychodzi z sieci — i dla partycji 2 wybierany jest nowy lider na brokerze 2.

Rys. 2. Broker 3 umiera, a jego follower na brokerze 2 zostaje nowym liderem partycji 2.

Następnie broker 1 również odchodzi, a partycja 1 traci swojego lidera, którego rolę przejmuje broker 2.

Rys. 3. Został tylko jeden broker. Wszyscy liderzy znajdują się na jednym brokerze z zerową nadmiarowością.

Gdy broker 1 powraca do sieci, dodaje czterech followerów, zapewniając pewną nadmiarowość dla każdej partycji. Ale wszyscy liderzy nadal pozostają na brokerze 2.

Rys. 4. Liderzy pozostają na brokerze 2.

Gdy broker 3 zostaje uruchomiony, wracamy do trzech replik w partycji. Ale wszyscy liderzy nadal są na brokerze 2.

Rys. 5. Niezbalansowane rozmieszczenie liderów po przywróceniu brokerów 1 i 3.

Kafka ma narzędzie do lepszego przekształcania liderów niż RabbitMQ. Tam trzeba było używać dodatkowych wtyczek lub skryptów, które zmieniały polityki na migrację głównego węzła, zmniejszając nadmiarowość w trakcie migracji. Ponadto przy dużych kolejach trzeba było pogodzić się z niedostępnością w czasie synchronizacji.

Kafka ma koncepcję 'preferowanych replik' na leadera. Gdy tworzone są partycje tematu, Kafka stara się równomiernie rozłożyć liderów po węzłach i oznacza tych pierwszych liderów jako preferowanych. Z czasem, z powodu restartów serwerów, awarii i problemów z łącznością, liderzy mogą znaleźć się na innych węzłach, jak w opisanym powyżej skrajnym przypadku.

Aby to naprawić, Kafka oferuje dwie opcje:

• Opcja auto.leader.rebalance.enable=true pozwala węzłowi kontrolerowi automatycznie przypisać liderów z powrotem do preferowanych replik, przywracając w ten sposób równomierne rozłożenie.
• Administrator może uruchomić skrypt kafka-preferred-replica-election.sh aby przypisać ręcznie.

Rys. 6. Repliki po rebalance

To była uproszczona wersja awarii, ale rzeczywistość jest bardziej skomplikowana, chociaż nie ma tutaj nic zbyt trudnego. Wszystko sprowadza się do zsynchronizowanych replik (In-Sync Replicas, ISR).

Zsynchronizowane repliki (ISR)

ISR to zestaw replik partycji, który jest uważany za 'zsynchronizowany'. Jest tu lider, ale mogą nie być żadne followery. Follower uznawany jest za zsynchronizowany, jeśli dokładnie skopiował wszystkie wiadomości lidera przed upływem interwału replica.lag.time.max.ms.

Follower jest usuwany z zestawu ISR, jeśli:

• nie złożył zapytania o pobranie w ciągu interwału replica.lag.time.max.ms (uznawany za martwego)
• nie zdążył zaktualizować się w ciągu interwału replica.lag.time.max.ms (uznawany za wolnego)

Followery składają zapytania o pobranie w interwale replica.fetch.wait.max.ms, który domyślnie wynosi 500 ms.

Aby wyraźnie wyjaśnić cel ISR, należy spojrzeć na potwierdzenia od producenta i niektóre scenariusze awarii. Producenci mogą zdecydować, kiedy broker wysyła potwierdzenie:

• acks=0, potwierdzenie nie jest wysyłane
• acks=1, potwierdzenie jest wysyłane po tym, jak lider zapisze wiadomość w swoim lokalnym dzienniku
• acks=all, potwierdzenie jest wysyłane po tym, jak wszystkie repliki w ISR zapisały wiadomość w lokalnych dziennikach

W terminologii Kafka, jeśli ISR zachowało wiadomość, następuje jej 'commit'. Acks=all to najbezpieczniejsza opcja, ale wiąże się z dodatkowym opóźnieniem. Rozważmy dwa przykłady awarii i jak różne opcje ‚acks’ wchodzą w interakcję z koncepcją ISR.

Acks=1 i ISR

W tym przykładzie zobaczymy, że jeśli lider nie oczekuje na potwierdzenie każdej wiadomości od wszystkich obserwatorów, to przy awarii lidera może dojść do utraty danych. Przejście do niesynchronizowanego obserwatora może być dozwolone lub zabronione w ustawieniach. unclean.leader.election.enable.

W tym przykładzie producent ustawia wartość acks=1. Partycja jest rozłożona pomiędzy wszystkie trzy brokerów. Broker 3 jest opóźniony, zsynchronizował się z liderem osiem sekund temu i obecnie ma 7456 wiadomości opóźnienia. Broker 1 opóźnił się o zaledwie jedną sekundę. Nasz producent wysyła wiadomość i szybko otrzymuje z powrotem ack, bez obciążenia ze strony wolnych lub martwych obserwatorów, których lider nie oczekuje.

Rys. 7. ISR z trzema replikami

Broker 2 ulega awarii, a producent otrzymuje błąd połączenia. Po przejściu liderstwa do brokera 1 tracimy 123 wiadomości. Obserwator na brokerze 1 był w ISR, ale nie był w pełni zsynchronizowany z liderem, gdy ten uległ awarii.

Rys. 8. Podczas awarii tracone są wiadomości

W konfiguracji bootstrap.servers producent wymienia kilku brokerów i może zapytać innego brokera, kto stał się nowym liderem partycji. Następnie nawiązuje połączenie z brokerem 1 i kontynuuje wysyłanie wiadomości.

Rys. 9. Wysyłanie wiadomości wznawiane po krótkiej przerwie

Broker 3 opóźnia się jeszcze bardziej. Wysyła żądania pobrania, ale nie może się zsynchronizować. Może to być spowodowane wolnym połączeniem sieciowym między brokerami, problemami z przechowywaniem itp. Zostaje usunięty z ISR. Teraz ISR składa się z jednej repliki — lidera! Producent dalej wysyła wiadomości i otrzymuje potwierdzenia.

Rys. 10. Obserwator na brokerze 3 usunięty z ISR

Broker 1 ulega awarii, a rola lidera przechodzi do brokera 3 z utratą 15286 wiadomości! Producent otrzymuje błąd połączenia. Przejście do lidera poza ISR było możliwe tylko dzięki ustawieniu unclean.leader.election.enable=true. Jeśli jest ustawione na false, to przejście by nie miało miejsca, a wszystkie żądania odczytu i zapisu zostałyby odrzucone. W takim przypadku czekamy na powrót brokera 1 z jego nietkniętymi danymi w replikacji, która ponownie przejmie liderstwo.

Rys. 11. Broker 1 ulega awarii. Podczas awarii traci się dużą ilość wiadomości

Producent łączy się z ostatnim brokerem i zauważa, że ten jest teraz liderem sekcji. Zaczyna wysyłać wiadomości do brokera 3.

Rys. 12. Po krótkiej przerwie wiadomości znowu są wysyłane do sekcji 0.

Zauważyliśmy, że oprócz krótkich przerw na ustanowienie nowych połączeń i szukanie nowego lidera, producent nieprzerwanie wysyłał wiadomości. Taka konfiguracja zapewnia dostępność kosztem spójności (bezpieczeństwa danych). Kafka utraciła tysiące wiadomości, ale nadal przyjmowała nowe wpisy.

Acks=all i ISR

Powtórzmy ten scenariusz jeszcze raz, ale z acks=all. Opóźnienie brokera 3 wynosi średnio cztery sekundy. Producent wysyła wiadomość z acks=all, a teraz nie dostaje szybkiej odpowiedzi. Lider czeka, aż wiadomość zostanie zapisana przez wszystkie repliki w ISR.

Rys. 13. ISR z trzema replikami. Jedna działa wolno, co prowadzi do opóźnienia zapisu.

Po czterech sekundach dodatkowego opóźnienia broker 2 wysyła ack. Wszystkie repliki są teraz w pełni zaktualizowane.

Rys. 14. Wszystkie repliki przechowują wiadomości i wysyłają ack.

Broker 3 teraz jeszcze bardziej się opóźnia i zostaje usunięty z ISR. Opóźnienie znacznie się zmniejsza, ponieważ w ISR nie ma już wolnych replik. Broker 2 czeka teraz tylko na brokera 1, który ma średnie opóźnienie 500 ms.

Rys. 15. Replika na brokerze 3 zostaje usunięta z ISR.

Następnie broker 2 awaryjnie kończy pracę, a liderstwo przechodzi do brokera 1 bez utraty wiadomości.

Rys. 16. Broker 2 awaryjnie kończy pracę.

Producent znajduje nowego lidera i zaczyna wysyłać mu wiadomości. Opóźnienie znowu się zmniejsza, ponieważ teraz ISR składa się z jednej repliki! Dlatego opcja acks=all nie dodaje nadmiarowości.

Rys. 17. Replika na brokerze 1 przejmuje liderstwo bez utraty wiadomości.

Następnie broker 1 awaryjnie kończy pracę, a liderstwo przechodzi do brokera 3 z utratą 14238 wiadomości!

Rys. 18. Broker 1 umiera, a przejście liderstwa z ustawieniem unclean prowadzi do znacznej utraty danych.

Nie musielibyśmy ustawiać opcji unclean.leader.election.enable na wartość true. Domyślnie wynosi ona false. Ustawienie acks=all z unclean.leader.election.enable=true zapewnia dostępność z pewnym dodatkowym bezpieczeństwem danych. Ale jak widzisz, nadal możemy stracić wiadomości.

Ale co, jeśli chcemy zwiększyć bezpieczeństwo danych? Można ustawić unclean.leader.election.enable = false, ale to niekoniecznie ochroni nas przed utratą danych. Jeśli lider mocno upadnie i zabierze ze sobą dane, wiadomości i tak będą utracone, a dostępność będzie stracona, dopóki administrator nie przywróci sytuacji do normy.

Lepiej zagwarantować nadmiarowość wszystkich wiadomości, w przeciwnym razie zrezygnować z zapisu. Wtedy przynajmniej z perspektywy brokera utrata danych może nastąpić tylko w przypadku dwóch lub więcej jednoczesnych awarii.

Acks=all, min.insync.replicas i ISR

Z konfiguracją topika min.insync.replicas zwiększamy poziom bezpieczeństwa danych. Przejdźmy jeszcze raz przez ostatnią część poprzedniego scenariusza, ale tym razem z min.insync.replicas=2.

Zatem broker 2 ma lidera repliki, a follower na brokerze 3 został usunięty z ISR.

Rys. 19. ISR z dwóch replik

Broker 2 upada, a liderstwo przechodzi do brokera 1 bez utraty wiadomości. Ale teraz ISR składa się tylko z jednej repliki. To nie spełnia minimalnej liczby do zapisów, i dlatego broker odpowiada na próbę zapisu błędem. NotEnoughReplicas.

Rys. 20. Liczba ISR o jeden niższa niż wskazana w min.insync.replicas

Ta konfiguracja poświęca dostępność na rzecz spójności. Zanim potwierdzimy wiadomość, zapewniamy, że jest zapisywana przynajmniej na dwóch replikach. To daje producentowi znacznie większą pewność. Tutaj utrata wiadomości jest możliwa tylko w przypadku jednoczesnego upadku dwóch replik w krótkim czasie, zanim wiadomość zostanie zreplikowana na dodatkowego followera, co jest mało prawdopodobne. Ale jeśli jesteś super paranoikiem, możesz ustawić współczynnik replikacji na 5, a min.insync.replicas na 3. Wtedy jednocześnie muszą upaść trzy brokery, aby utracić zapis! Oczywiście za taką niezawodność zapłacisz dodatkowym opóźnieniem.

Gdy dostępność jest konieczna dla bezpieczeństwa danych

Jak w w przypadku RabbitMQ, czasami dostępność jest konieczna dla bezpieczeństwa danych. Musisz pomyśleć o tym:

• Czy wydawca może po prostu zwrócić błąd, a wyższa usługa lub użytkownik spróbować później?
• Czy wydawca może zapisać wiadomość lokalnie lub w bazie danych, aby spróbować ponownie później?

Jeśli odpowiedź brzmi nie, optymalizacja dostępności zwiększa bezpieczeństwo danych. Utracisz mniej danych, jeśli wybierzesz dostępność zamiast rezygnacji z zapisu. Tak więc wszystko sprowadza się do znalezienia równowagi, a decyzja zależy od konkretnej sytuacji.

Sens ISR

Zestaw ISR pozwala wybrać optymalną równowagę między bezpieczeństwem danych a opóźnieniem. Na przykład zapewnia dostępność w przypadku awarii większości replik, minimalizując wpływ martwych lub wolnych replik z perspektywy opóźnienia.

Sami wybieramy wartość replica.lag.time.max.ms zgodnie z naszymi potrzebami. W zasadzie ten parametr oznacza, jakie opóźnienie jesteśmy gotowi zaakceptować przy acks=all. Wartość domyślna wynosi dziesięć sekund. Jeśli to dla Ciebie zbyt długo, możesz ją zmniejszyć. W takim przypadku wzrośnie częstość zmian w ISR, ponieważ followery będą częściej usuwane i dodawane.

W RabbitMQ to po prostu zestaw lustrzanych replik, które należy zreplikować. Wolne lustra wprowadzają dodatkowe opóźnienie, a odpowiedzi martwych luster można oczekiwać aż do wygaśnięcia czasu życia pakietów, które sprawdzają dostępność każdego węzła (net tick). ISR to interesujący sposób na uniknięcie tych problemów z wydłużonym opóźnieniem. Jednak ryzykujemy utratę nadmiarowości, ponieważ ISR może skrócić się tylko do lidera. Aby temu zapobiec, użyj ustawienia min.insync.replicas.

Gwarancja połączenia klientów

W ustawieniach bootstrap.servers producenta i konsumenta można wskazać kilku brokerów do połączenia z klientami. Pomysł polega na tym, że w przypadku awarii jednego węzła pozostaje kilka rezerwowych, z którymi klient może nawiązać połączenie. To niekoniecznie liderzy partycji, a po prostu punkt wyjściowy dla wstępnego załadowania. Klient może zapytać ich, na którym węźle znajduje się lider partycji do odczytu/zapisu.

W RabbitMQ klienci mogą łączyć się z dowolnym węzłem, a wewnętrzne routowanie kieruje zapytanie tam, gdzie trzeba. Oznacza to, że możesz ustawić przed RabbitMQ równoważnik obciążenia. Kafka wymaga, aby klienci łączyli się z węzłem, na którym znajduje się lider odpowiadającej partycji. W takiej sytuacji równoważnik obciążenia nie może być ustawiony. Lista bootstrap.servers jest krytycznie ważna, aby klienci mogli kierować się do odpowiednich węzłów i znajdować je po awarii.

Architektura konsensusu Kafka

Do tej pory nie omówiliśmy, jak klaster dowiaduje się o awarii brokera i jak wybierany jest nowy lider. Aby zrozumieć, jak Kafka radzi sobie z podziałami sieciowymi, najpierw należy zrozumieć architekturę konsensusu.

Każdy klaster Kafka jest uruchamiany wraz z klastrem Zookeeper — to usługa rozproszonego konsensusu, która pozwala systemowi osiągnąć konsensus w określonym stanie, kładąc priorytet na spójność nad dostępnością. Aby zatwierdzić operacje odczytu i zapisu, wymagane jest uzyskanie zgody większości węzłów Zookeeper.

Zookeeper przechowuje stan klastra:

• Listę tematów, partycji, konfiguracji, aktualne repliki lidera, preferowane repliki.
• Członkowie klastra. Każdy broker pinguje klaster Zookeeper. Jeśli nie otrzyma pinga w określonym czasie, Zookeeper uznaje brokera za niedostępnego.
• Wybór głównych i zapasowych węzłów dla kontrolera.

Węzeł kontrolera to jeden z brokerów Kafka, który odpowiada za wybór liderów replik. Zookeeper wysyła kontrolerowi powiadomienia o członkostwie w klastrze i zmianach w tematach, a kontroler musi działać zgodnie z tymi zmianami.

Na przykład, weźmy nowy temat z dziesięcioma partiami i współczynnikiem replikacji 3. Kontroler musi wybrać lidera dla każdej partii, starając się optymalnie rozłożyć liderów pomiędzy brokerami.

Dla każdej partii kontroler:

• aktualizuje informacje w Zookeeper na temat ISR i lidera;
• wysyła polecenie LeaderAndISRCommand do każdego brokera, który przechowuje replikę tej partii, informując brokerów o ISR i liderze.

Gdy broker z liderem ulega awarii, Zookeeper wysyła powiadomienie do kontrolera, który wybiera nowego lidera. Ponownie, kontroler najpierw aktualizuje Zookeeper, a następnie wysyła polecenie do każdego brokera, informując ich o zmianie w przywództwie.

Każdy lider jest odpowiedzialny za zbiór ISR. Konfiguracja replica.lag.time.max.ms określa, kto tam wejdzie. Gdy zmienia się ISR, lider przekazuje Zookeeper nowe informacje.

Zookeeper jest zawsze informowany o jakichkolwiek zmianach, aby w przypadku awarii przywództwo płynnie przeszło do nowego lidera.

Rys. 21. Konsensus Kafka

Protokół replikacji

Zrozumienie szczegółów replikacji pomaga lepiej zrozumieć potencjalne scenariusze utraty danych.

Zapytania o pobieranie, Log End Offset (LEO) i Highwater Mark (HW)

Rozpatrzyliśmy, że followerzy okresowo wysyłają liderowi zapytania o pobranie (fetch). Domyślny interwał wynosi 500 ms. Różni się to od RabbitMQ, w którym replikacja jest inicjowana przez mastera, a nie przez lustro kolejki. Master wysyła zmiany do luster.

Lider i wszyscy followerzy przechowują offset końca logu (Log End Offset, LEO) oraz znacznik Highwater (HW). Znacznik LEO przechowuje offset ostatniej wiadomości w lokalnej replicie, a HW — offset ostatniego komitu. Pamiętaj, że aby status „komit” mógł zostać uznany, wiadomość musi być zapisana we wszystkich replikach ISR. Oznacza to, że LEO zazwyczaj nieznacznie wyprzedza HW.

Gdy lider otrzymuje wiadomość, zapisuje ją lokalnie. Follower wysyła zapytanie o pobranie, przekazując swoje LEO. Następnie lider wysyła pakiet wiadomości, zaczynając od tego LEO, a także przesyła aktualne HW. Gdy lider uzyskuje informację, że wszystkie repliki zapisały wiadomość z danym offsetem, przesuwa znacznik HW. Tylko lider może przesunąć HW, a więc wszyscy followerzy poznają aktualną wartość w odpowiedziach na swoje zapytania. Oznacza to, że followerzy mogą pozostawać w tyle za liderem zarówno w kwestii wiadomości, jak i znajomości HW. Konsumenci otrzymują wiadomości tylko do aktualnego HW.

Zauważ, że „zapisany” (persisted) oznacza zapisany w pamięci, a nie na dysku. Dla wydajności Kafka synchronizuje dane na dysk w określonym interwale. RabbitMQ także ma taki interwał, ale potwierdzi publikację dopiero po zapisaniu wiadomości na dysku przez mastera i wszystkie lustra. Deweloperzy Kafki, ze względów wydajnościowych, zdecydowali się na wysyłanie ack, zaraz po zapisaniu wiadomości w pamięci. Kafka zakłada, że nadmiarowość zrekompensuje ryzyko krótkoterminowego przechowywania potwierdzonych wiadomości tylko w pamięci.

Awaria lidera

Gdy lider ulegnie awarii, Zookeeper informuje kontrolera, a ten wybiera nową replikę lidera. Nowy lider ustawia nowy znacznik HW zgodnie ze swoim LEO. Następnie followerzy otrzymują informację o nowym liderze. W zależności od wersji Kafki, follower wybierze jeden z dwóch scenariuszy:

1. Skróci lokalny log do znanego HW i wyśle do nowego lidera zapytanie o wiadomości po tym znaczniku.
2. Wyśle zapytanie do lidera, aby uzyskać HW w momencie jego wyboru na lidera, a następnie obetnie log do tego offsetu. Następnie zacznie wykonywać okresowe zapytania o próbki, zaczynając od tego offsetu.

Obserwującemu może być potrzebne obcięcie logu z następujących powodów:

• Gdy następuje awaria lidera, pierwszy obserwujący z zestawu ISR, zarejestrowany w Zookeeper, wygrywa wybory i staje się liderem. Wszyscy obserwujący w ISR, chociaż uznawani za 'zsynchronizowanych', mogli nie otrzymać od byłego lidera kopii wszystkich wiadomości. Może się zdarzyć, że wybrany obserwujący nie ma najnowszej kopii. Kafka gwarantuje, że między replikami nie ma rozbieżności. Dlatego, aby uniknąć rozbieżności, każdy obserwujący musi obciąć swój log do wartości HW nowego lidera w momencie jego wyboru. To kolejny powód, dla którego konfiguracja acks=all jest tak ważna dla spójności.
• Wiadomości są okresowo zapisywane na dysku. Jeśli wszystkie węzły klastra zawiodą jednocześnie, na dyskach pozostaną repliki o różnych offsetach. Może się zdarzyć, że gdy brokerzy znów wejdą do sieci, nowy lider, który zostanie wybrany, znajdzie się za swoimi obserwującymi, ponieważ zapisał się na dysk wcześniej niż inni.

Ponowne połączenie z klastrem

Podczas ponownego połączenia z klastrem repliki zachowują się tak samo jak przy awarii lidera: sprawdzają replikę lidera i obcinają swój log do jego HW (w momencie wyboru). Dla porównania, RabbitMQ traktuje ponownie podłączone węzły jako zupełnie nowe. W obu przypadkach broker odrzuca jakikolwiek istniejący stan. Jeśli używana jest automatyczna synchronizacja, mistrz musi zreplikować absolutnie całą bieżącą zawartość do nowego lustra w trybie 'niech cały świat poczeka'. Podczas tej operacji mistrz nie akceptuje żadnych operacji odczytu ani zapisu. Takie podejście powoduje problemy w dużych kolejkach.

Kafka to rozproszony dziennik, który generalnie przechowuje więcej wiadomości niż kolejka RabbitMQ, w której dane są usuwane z kolejki po ich odczytaniu. Aktywne kolejki powinny pozostawać stosunkowo małe. Jednak Kafka to dziennik z własną polityką przechowywania, która może ustalać okres w dniach lub tygodniach. Podejście polegające na blokowaniu kolejki i pełnej synchronizacji jest absolutnie niedopuszczalne dla rozproszonego dziennika. Zamiast tego, followery Kafki po prostu przycinają swój dziennik do HW lidera (w momencie jego wyboru), jeśli ich kopia wyprzedza lidera. W bardziej prawdopodobnym przypadku, gdy follower jest w tyle, po prostu zaczyna wysyłać zapytania o pobranie, zaczynając od swojego obecnego LEO.

Nowe lub dołączone followery zaczynają poza ISR i nie biorą udziału w zatwierdzeniach. Po prostu działają obok grupy, odbierając wiadomości tak szybko, jak to możliwe, aż dogonią lidera i wejdą do ISR. Nie ma tu blokady i nie ma potrzeby odrzucania wszystkich swoich danych.

Naruszenie spójności

Kafka ma więcej komponentów niż RabbitMQ, dlatego w tym przypadku zestaw zachowań jest bardziej skomplikowany, gdy w klastrze występują problemy z łącznością. Jednak Kafka została pierwotnie zaprojektowana z myślą o klastrach, więc rozwiązania są bardzo dobrze przemyślane.

Poniżej przedstawiono kilka scenariuszy naruszenia łączności:

• Scenariusz 1. Follower nie widzi lidera, ale nadal widzi Zookeeper.
• Scenariusz 2. Lider nie widzi żadnego followera, ale nadal widzi Zookeeper.
• Scenariusz 3. Follower widzi lidera, ale nie widzi Zookeeper.
• Scenariusz 4. Lider widzi followerów, ale nie widzi Zookeeper.
• Scenariusz 5. Follower jest całkowicie odizolowany od innych węzłów Kafki oraz od Zookeeper.
• Scenariusz 6. Lider jest całkowicie odizolowany od innych węzłów Kafki oraz od Zookeeper.
• Scenariusz 7. Węzeł kontrolera Kafki nie widzi innego węzła Kafki.
• Scenariusz 8. Kontroler Kafki nie widzi Zookeeper.

Dla każdego scenariusza przewidziane jest specjalne zachowanie.

Scenariusz 1. Follower nie widzi lidera, ale nadal widzi Zookeeper

Rys. 22. Scenariusz 1. ISR z trzech replik

Naruszenie łączności odcina brokera 3 od brokerów 1 i 2, ale nie od Zookeeper. Broker 3 nie może już wysyłać żądań o pobranie. Po upływie czasu replica.lag.time.max.ms jest usuwany z ISR i nie bierze udziału w commitach. Gdy tylko spójność zostanie przywrócona, wznowi zapytania o pobranie i dołączy do ISR, gdy dogoni lidera. Zookeeper będzie nadal odbierał pingi i uznawał, że broker jest żywy i zdrowy.

Rys. 23. Scenariusz 1. Broker jest usuwany z ISR, jeśli nie otrzyma zapytania o pobranie w czasie interwału replica.lag.time.max.ms

Nie ma żadnego logicznego podziału (split-brain) ani wstrzymania węzła, jak w RabbitMQ. Zamiast tego zmniejsza się nadmiarowość.

Scenariusz 2. Lider nie widzi żadnego followera, ale nadal widzi Zookeepera

Rys. 24. Scenariusz 2. Lider oraz dwóch followerów

Zaburzenie łączności sieciowej oddziela lidera od followerów, ale broker nadal widzi Zookeepera. Jak w pierwszym scenariuszu, ISR się kurczy, ale tym razem tylko do lidera, ponieważ wszyscy followerzy przestają wysyłać zapytania o pobranie. Ponownie, nie ma żadnego logicznego podziału. Zamiast tego następuje utrata nadmiarowości dla nowych wiadomości, dopóki spójność nie zostanie przywrócona. Zookeeper nadal odbiera pingi i uznaje, że broker jest żywy i zdrowy.

Rys. 25. Scenariusz 2. ISR skurczył się tylko do lidera

Scenariusz 3. Follower widzi lidera, ale nie widzi Zookeepera

Follower jest oddzielony od Zookeepera, ale nie od brokera z liderem. W efekcie follower nadal wysyła zapytania o pobranie i jest członkiem ISR. Zookeeper przestaje otrzymywać pingi i rejestruje awarię brokera, ale ponieważ to tylko follower, po przywróceniu nie ma żadnych konsekwencji.

Rys. 26. Scenariusz 3. Follower nadal wysyła zapytania o pobranie do lidera

Scenariusz 4. Lider widzi followerów, ale nie widzi Zookeepera

Rys. 27. Scenariusz 4. Lider oraz dwóch followerów

Lider jest oddzielony od Zookeepera, ale nie od brokerów z followerami.

Rys. 28. Scenariusz 4. Lider jest izolowany od Zookeepera

Po pewnym czasie Zookeeper zarejestruje awarię brokera i powiadomi o tym kontrolera. Ten wybierze nowego lidera spośród followerów. Jednak oryginalny lider nadal będzie uważał, że jest liderem i będzie kontynuował przyjmowanie zapisów z acks=1. Followerzy już mu nie wysyłają zapytań o pobranie, więc uzna ich za martwych i spróbuje skurczyć ISR do samego siebie. Ale ponieważ nie ma połączenia z Zookeeperem, nie będzie w stanie tego zrobić, a w tym momencie zrezygnuje z dalszego przyjmowania zapisów.

Wiadomości acks=all nie otrzymają potwierdzenia, ponieważ najpierw ISR obejmuje wszystkie repliki, a wiadomości do nich nie docierają. Kiedy pierwotny lider spróbuje usunąć je z ISR, nie będzie w stanie tego zrobić i w ogóle przestanie przyjmować jakiekolwiek wiadomości.

Klienci wkrótce zauważają zmianę lidera i zaczynają wysyłać zapisy na nowy serwer. Gdy sieć zostaje przywrócona, pierwotny lider widzi, że nie jest już liderem, i tnę swój log do wartości HW, którą miał nowy lider w momencie awarii, aby uniknąć rozbieżności logów. Następnie zacznie wysyłać zapytania do nowego lidera. Wszystkie zapisy pierwotnego lidera, które nie zostały zreplikowane do nowego lidera, zostaną utracone. Oznacza to, że wiadomości, które nie zostały potwierdzone przez pierwotnego lidera w te kilka sekund, gdy działało dwóch liderów, zostaną utracone.

Rys. 29. Scenariusz 4. Lider na brokerze 1 staje się obserwatorem po przywróceniu sieci.

Scenariusz 5. Obserwator jest całkowicie odizolowany zarówno od innych węzłów Kafka, jak i od Zookeepra.

Obserwator jest całkowicie izolowany zarówno od innych węzłów Kafka, jak i od Zookeepra. Po prostu zostaje usunięty z ISR, aż sieć zostanie przywrócona, a potem dogania pozostałych.

Rys. 30. Scenariusz 5. Izolowany obserwator zostaje usunięty z ISR.

Scenariusz 6. Lider jest całkowicie odizolowany zarówno od innych węzłów Kafka, jak i od Zookeepra.

Rys. 31. Scenariusz 6. Lider i dwóch obserwatorów.

Lider jest całkowicie odizolowany od swoich obserwatorów, kontrolera i Zookeepra. Przez krótki okres będzie nadal przyjmować zapisy z acks=1.

Rys. 32. Scenariusz 6. Izolacja lidera od innych węzłów Kafka i Zookeepra.

Nie otrzymując zapytań po upływie replica.lag.time.max.ms, spróbuje skompresować ISR do samego siebie, ale nie będzie w stanie tego zrobić, ponieważ nie ma połączenia z Zookeperem, wtedy przestanie przyjmować zapisy.

W międzyczasie Zookeeper oznaczy izolowanego brokera jako martwego, a kontroler wybierze nowego lidera.

Rys. 33. Scenariusz 6. Dwóch liderów.

Pierwotny lider może przyjmować zapisy przez kilka sekund, ale następnie przestaje przyjmować jakiekolwiek wiadomości. Klienci aktualizują się co 60 sekund z najnowszymi metadanymi. Zostaną poinformowani o zmianie lidera i zaczną wysyłać zapisy do nowego lidera.

Rys. 34. Scenariusz 6. Producenci przełączają się do nowego lidera.

Wszystkie potwierdzone zapisy dokonane przez pierwotnego lidera od momentu utraty spójności zostaną utracone. Gdy sieć zostanie przywrócona, pierwotny lider przez Zookeeper odkryje, że już nie jest liderem. Następnie przytnije swój dziennik do HW nowego lidera w momencie wyboru i zacznie wysyłać zapytania jako follower.

Rys. 35. Scenariusz 6. Pierwotny lider staje się followerem po przywróceniu spójności sieci

W tej sytuacji przez krótki czas może występować logiczny podział, ale tylko jeśli acks=1 i min.insync.replicas również 1. Logicznym podział kończy się automatycznie albo po przywróceniu sieci, gdy pierwotny lider zdaje sobie sprawę, że już nie jest liderem, albo gdy wszyscy klienci zrozumieją, że lider się zmienił i zaczną pisać do nowego lidera - w zależności od tego, co wydarzy się wcześniej. W każdym razie dojdzie do utraty niektórych wiadomości, ale tylko z acks=1.

Istnieje inna wersja tego scenariusza, w której tuż przed podziałem sieci followerzy zostają w tyle, a lider zmniejsza ISR do jednego siebie. Następnie izoluje się z powodu utraty spójności. Wybierany jest nowy lider, ale pierwotny lider nadal przyjmuje zapisy, nawet acks=all, ponieważ w ISR nie ma nikogo oprócz niego. Te zapisy zostaną utracone po przywróceniu sieci. Jedynym sposobem na uniknięcie takiej sytuacji jest min.insync.replicas = 2.

Scenariusz 7. Węzeł kontrolera Kafka nie widzi innego węzła Kafka

Ogólnie rzecz biorąc, po utracie połączenia z węzłem Kafka, kontroler nie będzie w stanie przekazać mu żadnych informacji dotyczących zmiany lidera. W najgorszym przypadku doprowadzi to do krótkoterminowego logicznego podziału, jak w scenariuszu 6. Najczęściej broker po prostu nie stanie się kandydatem na lidera w przypadku awarii ostatniego.

Scenariusz 8. Kontroler Kafka nie widzi Zookeepera

Od odłączonego kontrolera Zookeeper nie otrzyma pinga i wybierze nowy węzeł Kafka na kontrolerze. Pierwotny kontroler może nadal przedstawiać się jako taki, ale nie otrzymuje powiadomień od Zookeepera, więc nie ma żadnych zadań do wykonania. Gdy tylko sieć zostanie przywrócona, zrozumie, że już nie jest kontrolerem, lecz stał się zwykłym węzłem Kafka.

Wnioski z scenariuszy

Widoczne jest, że utrata łączności obserwatorów nie prowadzi do utraty wiadomości, a jedynie tymczasowo zmniejsza nadmiarowość, aż do momentu przywrócenia sieci. Może to oczywiście prowadzić do utraty danych, jeśli stracone zostaną jeden lub kilka węzłów.

Jeśli z powodu utraty łączności lider odłączy się od Zookeepera, może to prowadzić do utraty wiadomości z acks=1. Brak łączności z Zookeeperem powoduje krótkotrwały podział logiczny z dwoma liderami. Problem ten rozwiązuje parametr acks=all.

Parametr min.insync.replicas w dwóch lub więcej replikach zapewnia dodatkowe gwarancje, że takie krótkoterminowe scenariusze nie doprowadzą do utraty wiadomości, jak w scenariuszu 6.

Podsumowanie dotyczące utraty wiadomości

Wymieńmy wszystkie sposoby, w jakie można stracić dane w Kafka:

• Jakakolwiek awaria lidera, jeśli wiadomości były potwierdzane za pomocą acks=1
• Jakikolwiek nieczysty (unclean) transfer lidera, czyli na obserwatora poza ISR, nawet z acks=all
• Izolacja lidera od Zookeepera, jeśli wiadomości były potwierdzane za pomocą acks=1
• Pełna izolacja lidera, który już skurczył grupę ISR tylko do siebie. Zostaną utracone wszystkie wiadomości, nawet acks=all. To prawda tylko w przypadku, jeśli min.insync.replicas=1.
• Jednoczesne awarie wszystkich węzłów danego segmentu. Ponieważ wiadomości są potwierdzane z pamięci, niektóre mogą jeszcze nie zostać zapisane na dysku. Po ponownym uruchomieniu serwerów może brakować niektórych wiadomości.

Nieczystych transferów lidera można uniknąć, albo ich zabraniając, albo zapewniając nadmiarowość na poziomie co najmniej dwóch. Najbardziej solidna konfiguracja to połączenie acks=all i min.insync.replicas więcej niż 1.

Bezpośrednie porównanie niezawodności RabbitMQ i Kafka

Aby zapewnić niezawodność i wysoką dostępność, obie platformy wdrażają system replikacji pierwotnej i wtórnej. Jednak RabbitMQ ma swoje słabe miejsce. Po ponownym połączeniu po awarii węzły zrzucają swoje dane, a synchronizacja zostaje zablokowana. Ten podwójny cios podważa trwałość dużych kolejek w RabbitMQ. Będziesz musiał pogodzić się albo z ograniczeniem nadmiarowości, albo z długotrwałymi blokadami. Zmniejszenie nadmiarowości zwiększa ryzyko masowej utraty danych. Ale jeśli kolejki są małe, to w celu zapewnienia nadmiarowości z krótkimi okresami niedostępności (kilka sekund) można poradzić sobie z pomocą ponownych prób połączenia.

W Kafka nie ma takiego problemu. Odrzuca dane tylko w punkcie niezgodności lidera i obserwatora. Wszystkie wspólne dane są zachowywane. Ponadto replikacja nie blokuje systemu. Lider nadal przyjmuje wpisy, podczas gdy nowy obserwator go dogania, więc dla devopsów dołączanie lub ponowne łączenie klastra staje się trywialnym zadaniem. Oczywiście nadal istnieją problemy, takie jak przepustowość sieci podczas replikacji. Jeśli jednocześnie dodano kilku obserwatorów, można napotkać limit przepustowości.

RabbitMQ przewyższa Kafka w niezawodności w przypadku jednoczesnej awarii wielu serwerów w klastrze. Jak już wspomnieliśmy, RabbitMQ wysyła potwierdzenie do publikacji dopiero po zapisaniu wiadomości na dysku u lidera i wszystkich lustrzanych. Jednak wprowadza to dodatkowe opóźnienie z dwóch powodów:

• fsync co kilka setek milisekund
• Awarię lustra można zauważyć dopiero po upływie czasu życia pakietów, które sprawdzają dostępność każdego węzła (net tick). Jeśli lustro spowalnia lub pada, dodaje to opóźnienie.

Kafka stawia na to, że jeśli wiadomość jest przechowywana na kilku węzłach, można potwierdzać wiadomości, gdy tylko trafią do pamięci. Z tego powodu istnieje ryzyko utraty wiadomości wszelkiego rodzaju (nawet acks=all, min.insync.repliki=2) w przypadku jednoczesnej awarii.

Ogólnie rzecz biorąc, Kafka wykazuje wyższą wydajność i jest pierwotnie zaprojektowana dla klastrów. Liczbę obserwatorów można zwiększyć do 11, jeśli jest to potrzebne dla niezawodności. Współczynnik replikacji 5 i minimalna liczba replik w synchronizowanym stanie min.insync.replicas=3 sprawią, że utrata wiadomości stanie się bardzo rzadkim zjawiskiem. Jeśli twoja infrastruktura jest w stanie zapewnić taki współczynnik replikacji i poziom redundancji, możesz wybrać tę opcję.

Klastrowanie RabbitMQ jest dobre dla małych kolejek. Ale nawet małe kolejki mogą szybko urosnąć przy dużym ruchu. Gdy kolejki stają się duże, trzeba podjąć trudną decyzję między dostępnością a niezawodnością. Klastrowanie RabbitMQ najlepiej nadaje się do nietypowych sytuacji, gdzie zalety elastyczności RabbitMQ przewyższają wszelkie wady jego klastrowania.

Jednym z antidotum na lukę RabbitMQ w odniesieniu do dużych kolejek jest podział ich na wiele mniejszych. Jeśli nie wymagasz pełnego uporządkowania całej kolejki, a jedynie odpowiednich wiadomości (na przykład wiadomości od konkretnego klienta), lub w ogóle nie chcesz niczego porządkować, to taka opcja jest akceptowalna: zobacz mój projekt Rebalanser do podziału kolejki (projekt jest jeszcze na wczesnym etapie).

Na koniec nie zapominaj o szeregu błędów w mechanizmach klastrowania i replikacji zarówno RabbitMQ, jak i Kafka. Z czasem systemy stały się bardziej dojrzałe i stabilne, ale żadna wiadomość nigdy nie będzie w 100% chroniona przed utratą! Ponadto w centrach danych mogą wystąpić awarie na dużą skalę!

Jeśli coś przeoczyłem, zrobiłem błąd lub się nie zgadzasz z którymkolwiek z tez, nie wahaj się napisać komentarza lub skontaktować się ze mną.

Często pytają mnie: „Co wybrać, Kafka czy RabbitMQ?”, „Która platforma jest lepsza?”. Prawda jest taka, że to naprawdę zależy od Twojej sytuacji, obecnego doświadczenia itd. Nie chcę wyrażać swojego zdania, ponieważ byłoby zbyt dużym uproszczeniem polecać jakąś jedną platformę do wszystkich zastosowań i możliwych ograniczeń. Napisałem ten cykl artykułów, abyś mógł wyrobić sobie własną opinię.

Chcę powiedzieć, że oba systemy są liderami w tej dziedzinie. Może jestem trochę stronniczy, ponieważ na podstawie swoich projektów bardziej cenię takie rzeczy jak gwarantowane uporządkowanie wiadomości i niezawodność.

Widziałem inne technologie, którym brakuje tej niezawodności i gwarantowanego uporządkowania, a potem patrzę na RabbitMQ i Kafka — i dostrzegam niesamowitą wartość obu tych systemów.

Źródło: habr.com