Cześć! Witaj w kolejnym wpisie z serii Apache Kafka.

Zanim jednak do tego przejdę… Już za niedługo – bo w maju – startujemy z przedsprzedażą najbardziej kompleksowego Programu Szkoleniowego Java Developera w kraju 💪. Trzech doświadczonych programistów nad Programem pracuje ponad rok.

Dopinamy ostatnie kwestie techniczne, rendery materiałów, wprowadzamy poprawki po review i za kilka tygodni oddamy w Twoje ręce Program Java Developera 🚀

12 modułów udostępnianych tydzień po tygodniu, 40 godzin lekcji video, sporo fajnych technologii i DOŻYWOTNI dostęp do materiałów. Ten Program Szkoleniowy to prawdziwy SZTOS!

Pełną agendę znajdziesz TUTAJ 🔥
A już dzisiaj zapisać się możesz TUTAJ – damy Ci znać, kiedy startujemy i dostaniesz ofertę z najlepszą ceną ever!
Do rzeczy, co z tymi najczęstszymi błędami popełnianymi przy integracji za pomocą Apache Kafka!

Wstęp

Integracja aplikacji ze sobą za pomocą Apache Kafka jest stosunkowo łatwa. W tym zadaniu wspomagają nas biblioteki albo całe frameworki (takie jak Spring). Ponieważ to zadanie jest znacznie ułatwione, może to uśpić czujność programistów. Chciałbym zwrócić Twoją uwagę na to, żebyś poświęcił chwilę na analizę wymagań niefunkcjonalnych w swojej aplikacji, a także zapoznał się z domyślnymi ustawieniami Apache Kafka i bibliotek integrujących.

Apache Kafka to bardzo rozbudowany system kolejkowy, dający się dostosować do niemalże każdej potrzeby biznesowej. Zupełnie odmienne wymagania niefunkcjonalne będą stawiane w przypadku przelewów bankowych a inne przy obsłudze kolejkowania danych statystycznych do marketingu. Odpowiednie ustawienie parametrów pozwala Ci wybrać pomiędzy szybkością a gwarancją dostarczenia komunikacji.

Zadajesz sobie pewnie teraz pytanie – czy nie można tego ujednolicić i zaprojektować szybką, ale i bezpieczną wersję systemu kolejkowego? Otóż nie!
Przy tzw. „szybkiej wersji” producenta oraz konsumenta nie interesuje nas potwierdzenie odbioru wiadomości przez klaster Kafki. Szybki producent ma za zadanie wysyłanie wiadomości i nie oczekuje potwierdzenia odebrania od brokera. Repliki partycji nie muszą potwierdzać, że otrzymały wiadomości od producenta – to wszystko trwa pewien czas. Rezygnując z tego wymagania jesteś w stanie wysyłać dziesiątki milionów komunikatów na sekundę. Ale co się stanie jeśli producent wyśle wiadomość, nie oczekując przy tym na potwierdzenie odbioru z klastra Kafki, a nastąpiły problemy przy zapisie wiadomości na partycji? Istnieje ryzyko utraty wiadomości. Sytuacja jest niedopuszczalna, jeśli stracona wiadomość to informacja o przelewie w systemie bankowym. Z drugiej strony, jeśli byłaby to wiadomość czysto statystyczna, że użytkownik wykonał konkretną akcję na stronie i zbieramy takie informacje do marketingu – wówczas jeśli utracimy promil komunikacji – możliwe, że nie ma poważnych konsekwencji. W przypadku statystycznych danych potrzebujemy maksymalnej szybkości – nie zważając na spójność danych. W przypadku obsługi przelewów przez system kolejkowy – potrzebujemy maksymalnej spójności danych, akceptując większe opóźnienia.

Po stronie konsumenta wiadomości w systemie kolejkowym Apache Kafka również możemy wyróżnić „szybką” oraz „bezpieczną” wersję. Co ważne podkreślenia, domyślne ustawienia Apache Kafka to wersja szybka konsumenta. Jeśli chcemy zachować maksymalną spójność danych powinniśmy zastosować odpowiednie ustawienia i zrezygnować z automatycznego commitowania offsetu wiadomości, na rzecz manualnego commitu. Powinniśmy również odpowiednio reagować na proces rebalancingu i zanim aplikacja utraci przydział odczytu danej partycji jako konsument – wtedy należy zakomitować offset. Moment zakomitowania offsetu ma wpływ na to, czy w razie awarii utracimy komunikaty, czy odczytamy je ponownie. Szybciej jest tracić komunikaty i nigdy do nich nie wracać. Bezpieczniej jest odczytywać potencjalne duplikaty, ale sprawdzać dodatkowo, czy aby na pewno nie przerobiliśmy danych wiadomości – wymaga to dodatkowej obsługi po stronie konsumenta wiadomości oraz takie zaprojektowanie naszej aplikacji, aby była idempotentna. Przejdźmy w końcu do listy najpopularniejszych błędów obsługi kolejek Apache Kafka.

Błąd 1 – Użycie domyślnych ustawień producenta

Aby zacząć pisać wiadomości do systemu kolejkowego Apache Kafka – wystarczy zrobić kilka bardzo prostych konfiguracji:

• Skonfigurować klaster Kafki (broker, topic…)
• Do projektu dołączyć bibliotekę Apache Kafka

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.8.0</version>
</dependency>

• Ustawić takie parametry jak bootstrap.servers oraz serializator klucza i wartości wiadomości

Properties kafkaProps = new Properties();
kafkaProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
kafkaProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
kafkaProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

• W końcu możemy stworzyć instancję producenta oraz wysłać wiadomość, Kod w Java może wyglądać następująco

// create producer
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(kafkaProps);

// crate a producer records
ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key", "simple fire and forget producer");

// send data
producer.send(record);

//clear connection
producer.flush();
producer.close();

I ciach – wysłaliśmy właśnie event (wiadomość) na kolejkę Apache Kafka. Proste, prawda?

A teraz wyobraź sobie, że broker lub leader partycji ulega awarii. Komunikat nie zostaje fizycznie dostarczony do konsumenta albo zostaje dostarczony kilkukrotnie. Tak może skończyć się pozostanie przy domyślnych ustawieniach producenta.

Aby zapobiec lub bardzo ograniczyć takie niechciane działania należy wybrać odpowiedni typ producenta (synchroniczny, asynchroniczny, fire and forget) oraz zainteresować się takimi parametrami konfiguracyjnymi takimi jak:

• acks
• replication.factor
• unclean.leader.election.enable
• min.insync.replicas
• max.in.flight.request.per.connection
• retries.config
• enable.idempotence
• max.poll.records
• fetch.min.bytes
• fetch.max.wait.ms

Omówienie tych wszystkich konfiguracji znacznie wykracza poza zakres tego artykułu (czytałbyś go kilka dni:)) Jednak w Programie Szkoleniowym Java Developera poświęciliśmy 2 obszerne moduły poświęcone tematyce Apache Kafka (ponad 7h materiału i przykłady w kodzie). Dla przypomnienia:

Pełną agendę znajdziesz TUTAJ 🔥
A już dzisiaj zapisać się możesz TUTAJ 🔥

Błąd 2 – Użycie domyślnych ustawień konsumenta

Podobnie jak w przypadku producenta, podstawowa konfiguracja konsumenta umożliwiająca odczyt komunikatów jest bardzo prosta. Kod w Java może wyglądać następująco:

//1. create consumer properties
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

// 2. Create KafkaConsumer
KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(consumerProps);

// 3. Subscribe the topic
consumer.subscribe(Collections.singletonList(TOPIC_NAME));

//4. Poll records
try {
	while (true) { // bad practice while(true)
		ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
		for (ConsumerRecord<String, String> record : records) {
			log.info("topic = {}, partition = {}, offset = {}, key = {}, value = {}",
					record.topic(), record.partition(), record.offset(), record.key(), record.value());
			//doing something with records
		}
	}
} catch (Exception e) {
	log.error("Exception while poll messages", e);
} finally {
	// 5. Close the consumer
	consumer.close();
}

I ciach, parę minut i możemy czytać komunikaty z kolejki. Prostę, prawda?

A teraz wyobraź sobie, że w klastrze Kafki następuje proces rebalancingu. Nasz konsument traci przynależność do partycji, którą czytał do tej pory. Automatyczny proces commitowania offsetu (domyślne ustawienie) nie zdążył zacommitować offsetu przeprocesowanych wiadomości albo zrobił, to zanim zostały one fizycznie przeprocesowane. Wówczas albo stracimy komunikację, albo odczytamy ją kilkukrotnie. Również w przypadku konsumenta błędem jest poleganie na domyślnych ustawieniach klastra i naiwne myślenie, że nic się nie stanie. A widziałem już błędy na bankowych produkcjach, kiedy przelewy były księgowane podwójnie sic!

Aby zabezpieczyć się przed błędnym odczytem wiadomości należy zastanowić się nad manualnym commitowaniem offsetu wiadomości (w Kafce mamy 3 tryby commitu offsetu: automatyczny, manualny synchroniczny oraz manualny asynchroniczny). Ponadto warto jest prawidłowo (w zależności od potrzeby biznesowej) ustawić takie parametry jak:

• session.timeout.ms
• heartbeat.interval.ms
• max.poll.interval.ms
• max.poll.records

Dodatkowo trzeba odpowiednio zareagować na proces rebalancingu oraz w porę zacommitować offset przerobionych wiadomości.

Również tak jak w przypadku producenta, z tych informacji można napisać nie jedną książkę. Bezpieczeństwie oraz szybkości producenta i konsumenta poświęciliśmy ponad 3h materiału (cały moduł) w naszym Programie Szkoleniowym Java Developera.

Pełną agendę znajdziesz TUTAJ 🔥
A już dzisiaj zapisać się możesz TUTAJ 🔥

Błąd 3 – Brak idempotentności konsumenta

W poprzednich dwóch punktach przybliżyłem Ci temat bezpieczeństwa producenta oraz konsumenta. W systemach kolejkowych istnieje takie pojęcie jak exactly-once, czyli „dostarczaj tylko raz”. Jest naukowo udowodnione, że jest to zagadnienie teoretyczne i w praktycznym świecie nie ma możliwości uzyskania takiego trybu. Co prawda, poprzez odpowiedni kod i konfigurację klasta Kafki zbliżamy się do tego idealnego stanu, ale musimy mieć świadomość, że nigdy go nie osiągniemy (nawet jak z duplikujemy jedną wiadomość na milion odebranych). Dla tego w przypadku krytycznych procesów należy zapewnić, aby nasz konsument wiadomości był idempotentny. Czyli – aby powtórne odczytanie i prze-procesowanie komunikatu nie wyrządziło krzywdy naszym danym.

Błąd 4 – Ustawienie jednej partycji

Jedna partycja na topic to tak jak jednopasmowa droga zamiast 3 pasmowej autostrady. Bardzo ogranicza wydajność, ale wpływa też na bezpieczeństwo naszego całego klastra. Dla tego przy konfiguracji klastra warto jest ustawić co najmniej 2 partycje na jeden topic oraz pochylić się tematowi replikacji tych partycji. Tak, aby w przypadku utraty fizycznego pliku gdzie trzymane są komunikaty (partycja) – jej kopia (replika) podjęła ten wysiłek

Błąd 5 – Ustawienie jednej instancji brokera

Poleganie na jednej instancji brokera to jak trzymanie wszystkich jajek w jednym koszyku. Co się stanie, jeśli broker ulegnie awarii albo wystąpią problemy sieciowe na styku producent <-> broker <-> konsument? W przypadku świata rozproszonych mikroserwisów, to nie jest nic nadzwyczajnego (o mikroserwisach mamy 2 moduły w naszym Programie Szkoleniowym – symulujemy oraz pokazujemy strategie wyjścia z najpopularniejszych problemów). Dla tego, kiedy konfigurujesz klaster Kafki, stwórz co najmniej 3 instancje brokera!

Podsumowanie

Podsumowując: Kafka nie może być piekielnie szybka, jednocześnie oferując
maksymalną spójność danych. Do wybrania odpowiednich parametrów systemu
należy przeprowadzić analizę wymagań niefunkcjonalnych dla aplikacji. Domyślne
ustawienia oferują szybką pracę systemu kolejkowego w zamian za spójność
danych. Domyślne ustawienia będą dobre przy obsłudze danych, na których nie
specjalnie nam zależy. Jeśli jednak dane są bardzo istotne i nie możemy sobie
pozwolić na ich utratę czy duplikację – powinniśmy zmienić domyślne ustawienia
Apache Kafka oraz w odpowiedni sposób ustawić parametry producenta,
konsumenta wiadomości oraz klastra Kafki.

Wpis który czytasz to zaledwie fragment wiedzy zawartej w Programie szkoleniowym Java Developera od SoftwareSkill. Mamy do przekazania sporo usystematyzowanej wiedzy z zakresu kluczowych kompetencji i umiejętności Java Developera. Program składa się z kilku modułów w cotygodniowych dawkach wiedzy w formie video.

Gdybyś potrzebował jeszcze więcej: