Pregled ključnih koncepata Kafke

Osnovni uvod

Apache Kafka je otvorena platforma za obradu tokova podataka koju je LinkedIn razvio u Skali i Javi, a zatim je poklonio Apache Software Foundation-u.

Cilj projekta je da pruži jedinstvenu platformu visokog protoka i niske latencije za obradu tokova podataka u realnom vremenu.

Kafka se preko Kafka Connect-a može povezati sa spoljnim sistemima, a uz to pruža i Kafka Streams.

"Karakteristike Kafke"

Kafka je distribuirani sistem zasnovan na modelu objavljivanja/pretplate (publish/subscribe), a glavne karakteristike su:

Karakteristika	Distribuiranost	"Visoke performanse"	"Postojanost i skalabilnost"
Opis	Više particija	Visok protok	Podaci se mogu trajno čuvati
	Više replika	Niska latencija	Tolerancija na greške
	Više pretplatnika	Visoka konkurentnost	Podrška za horizontalno skaliranje
	Koordinacija preko ZooKeeper-a	Vremenska složenost O(1)	Automatsko balansiranje poruka

Verzija

"Imenovanje verzija Kafke"

Kada preuzimamo Kafku sa zvaničnog sajta, videćemo verziju u ovakvom obliku:

Prvi broj verzije odgovara verziji Scala kompajlera kojim se prevodi izvorni kod Kafke.

Kôd na strani Kafka servera u potpunosti je napisan u Scali. Scala istovremeno podržava objektno-orijentisano i funkcionalno programiranje, a izvorni kod napisan u Scali se nakon prevođenja pretvara u obične .class fajlove, zbog čega kažemo da Scala spada u porodicu jezika zasnovanih na JVM-u.

Pravi broj verzije Kafke zapravo je 2.1.1.

❝

Šta znači ovo 2.1.1?

❞

Prva cifra 2 označava glavnu verziju, odnosno Major Version; srednja cifra 1 označava minor verziju, odnosno Minor Version; poslednja cifra 1 označava revizijsku verziju, odnosno Patch broj.

Kafka zajednica je nakon izdavanja verzije 1.0.0 objavila članak kojim je saopštila da se pravilo imenovanja verzija Kafke zvanično prešlo sa četvorocifrenog na trocifreno, npr. verzija 0.11.0.0 bila je četvorocifrena.

Jedan savet: bez obzira na to koju verziju koristite, trudite se da verzija na strani servera i verzija na strani klijenta budu usaglašene, jer ćete u suprotnom izgubiti veliki deo performansi koje vam Kafka nudi.

"Evolucija verzija"

Verzija 0.7: nudila je samo najosnovniju funkcionalnost reda poruka.

Verzija 0.8: uveden je mehanizam replika, čime Kafka postaje zaista potpuno distribuirano i pouzdano rešenje za redove poruka.

Verzija 0.9.0.0: dodate su osnovne funkcije bezbednosne autentifikacije/ovlašćenja; novi Consumer API je ponovo napisan u Javi; uvedena je komponenta Kafka Connect.

Verzija 0.11.0.0: pružen je idempotentni Producer API i transakcioni API; izvršena je redizajn strukture formata poruka Kafke.

Verzije 1.0 i 2.0: uglavnom razne ispravke i unapređenja za Kafka Streams.

Osnovni pojmovi

"Tema (Topic)"

Objekat objavljivanja i pretplate je tema (Topic). Za svaki biznis, svaku aplikaciju, pa čak i za svaku vrstu podataka, može se napraviti posebna tema.

"Proizvođači i potrošači"

Klijentske aplikacije koje objavljuju poruke u temu nazivaju se proizvođačima. Programske komponente proizvođača obično neprekidno šalju poruke u jednu ili više tema.

Klijentske aplikacije koje se pretplate na poruke iz ovih tema nazivaju se potrošačima. Potrošači takođe mogu istovremeno da budu pretplaćeni na više tema.

"Broker"

Klaster čini više Broker-a. Broker je zadužen za prijem i obradu zahteva koje šalje klijent, kao i za trajno čuvanje poruka.

Iako više Broker procesa može da radi na istoj mašini, češća je praksa da se različiti Broker-i rasporede na različite mašine. Tako, ako jedna mašina u klasteru padne, čak i ako svi Broker procesi na njoj prekinu rad, Broker-i na ostalim mašinama i dalje mogu nesmetano da pružaju usluge.

"Mehanizam rezervnih kopija"

Ideja je jednostavna: isti podaci se kopiraju na više mašina, a takve identične kopije nazivaju se replikama.

Definisane su dve vrste replika: lider replika i pratilec replika.

Prva vrsta pruža usluge ka spolja, što ovde znači da komunicira sa klijentskim programima; druga vrlika samo pasivno prati lider repliku i ne može da komunicira sa spoljnim svetom.

"Particije"

Mehanizam particija znači da se svaka tema podeli na više particija, pri čemu je svaka particija jedan uređen dnevnik poruka.

Svaka poruka koju proizvede proizvođač biće poslata samo u jednu particiju. Drugim rečima, ako se u temu sa dve particije pošalje jedna poruka, ona će završiti ili u particiji 0 ili u particiji 1.

Ispod svake particije može se konfigurisati više replika, od kojih samo jedna može biti lider replika, a preostalih N-1 su pratilec replike.

Kada proizvođač upisuje poruke u particiju, pozicija svake poruke unutar particije naziva se ofsetom.

"Grupa potrošača"

Više instanci potrošača zajedno čine jednu grupu koja zajednički konzumira skup tema.

Svaka particija unutar tog skupa tema biće konzumirana od strane tačno jedne instance unutar grupe, dok ostale instance ne mogu da je konzumiraju.

❝

Na ovaj način se istovremeno ostvaruju oba velika modela tradicionalnih sistema za obradu poruka:

❞

Ako sve instance pripadaju istoj Group, realizuje se model reda poruka;

Ako sve instance pripadaju različitim Group-ama, realizuje se model objavljivanja/pretplate.

"Koordinator: Coordinator"

Koordinator je komponenta koja služi isključivo Consumer Group-i. Odgovorna je za izvršavanje Rebalance-a za tu grupu, kao i za upravljanje ofsetima i članovima grupe.

Konkretno, kada aplikacija na strani Consumer-a podnosi ofset, ona zapravo šalje ofset Broker-u na kom se nalazi Coordinator. Isto tako, kada se Consumer aplikacija pokrene, ona šalne razne zahteve ka Broker-u na kom se nalazi Coordinator, a zatim Coordinator izvršava operacije upravljanja metapodacima kao što su registracija grupe potrošača i vođenje evidencije o njenim članovima.

Svi Broker-i pri pokretanju kreiraju i aktiviraju odgovarajuću komponentu Coordinator.

Drugim rečima, "svaki Broker ima sopstvenu komponentu Coordinator".

Kako Consumer Group saznaje na kom Broker-u se nalazi Coordinator koji joj pruža usluge?

Preko interne teme Kafke __consumer_offsets.

Trenutno Kafka određuje Broker-a na kom se nalazi Coordinator za neku Consumer Group u dva koraka.

• Korak 1: određuje se koja particija teme __consumer_offsets čuva podatke te grupe: partitionId=Math.abs(groupId.hashCode() % offsetsTopicPartitionCount).
• Korak 2: pronalazi se Broker na kom se nalazi lider replika te particije; taj Broker je traženi Coordinator.

Najpre Kafka izračunava heš vrednost parametra group.id date grupe.

Na primer, ako grupi group.id postavite vrednost test-group, njen hashCode bi trebalo da bude 627841412.

Zatim Kafka određuje broj particija teme __consumer_offsets, što je obično 50, i zatim računa apsolutnu vrednost ostatka prethodnog heša pri deljenju sa brojem particija, odnosno abs(627841412 % 50) = 12.

Na taj način saznajemo da particija 12 teme __consumer_offsets čuva podatke o ovoj grupi.

Kada imamo broj particije, dovoljno je da pronađemo na kom Broker-u se nalazi lider replika particije 12 teme __consumer_offsets; taj Broker je traženi Coordinator.

"Ofset potrošača: Consumer Offset"

Napredak konzumiranja potrošača; svaki potrošač ima sopstveni ofset potrošača.

"Rebalans: Rebalance"

Proces u kom ostale instance potrošača automatski redistribuiraju particije pretplaćenih tema nakon što jedna instanca u grupi padne.

Rebalance je važan mehanizam kojim Kafka na strani potrošača ostvaruje visoku dostupnost.

"AR (Assigned Replicas)": sve replike unutar particike se zajednički nazivaju AR.

Sve poruke se prvo šalju lider repliki, a zatim pratilec replike mogu da povuku poruke od lidera radi sinhronizacije.

Međutim, tokom sinhronizacije, pratilec u odnosu na lider ima izvesno kašnjenje, pa u tom trenutku pratilec i lider nisu u potpunoj sinhronizaciji.

"OSR (Out Sync Replicas)": skup replika koje nisu u potpunoj sinhronizaciji sa liderom ili zaostaju za njom.

"ISR (In Sync Replicas):" "podskup AR-a; replike u ISR-u su" one koje su u potpunoj sinhronizaciji sa liderom. Ako neka pratilec replika u ISR-u previše zaostaje za liderom, biće uklonjena iz ISR-a; s druge strane, ako dostigne punu sinhronizaciju, prebačuje se iz OSR-a u ISR.

Po podrazumevanoj postavci, kada lider replika otkazuje, samo pratilec replike u ISR skupu imaju pravo da budu izabrane za novog lidera, dok replike iz OSR-a nemaju tu mogućnost (ovo se može konfigurisati preko unclean.leader.election.enable).

"HW (High Watermark)": oznaka visokog vodenog nivoa; ona označava specifičan ofset poruke, a potrošači mogu da povuku samo poruke pre tog ofseta.

Slika ispod prikazuje jedan log fajl u kom se nalazi 9 poruka. Prva poruka ima ofset (LogStartOffset) 0, poslednja ima ofset 8, dok je poruka sa ofsetom 9 prikazana isprekidanom linijom i predstavlja sledeću poruku koja tek treba da bude upisana.

Visoki vodeni nivo (HW) log fajla je 6, što znači da potrošač može da povuče samo poruke sa ofsetom od 0 do 5. Poruka sa ofsetom 6 za potrošača nije vidljiva.

"LEO (Log End Offset)": označava ofset sledeće poruke koja tek treba da bude upisana u trenutni log fajl.

Na gornjoj slici, pozicija sa ofsetom 9 je LEO trenutnog log fajla. Vrednost LEO-a jednaka je ofsetu poslednje poruke u trenutnoj particiji log-a uvećanom za 1.

Svaka replika u skupu ISR particije održava sopstveni LEO, a najmanji LEO u skupu ISR predstavlja HW particije; potrošač može da konzumira samo poruke do HW-a.

Sistemska arhitektura

"Principi dizajna Kafke"

Najosnovnija arhitektura je takva da proizvođač objavljuje poruku u jedan Kafka Topic; poruke te teme čuvaju se u Broker-u. Potrošač se pretplaćuje na tu temu, a zatim konzumira poruke iz Broker-a. Sledeća slika opisuje ovaj scenario:

"Stanje poruka:" U Kafci se stanje „da li je poruka konzumirana“ čuva na strani Consumer-a. Broker ne vodi računa o tome da li je neka poruka konzumirana niti ko ju je konzumirao. Consumer vodi evidenciju o ofsetu (pokazivač na sledeću poruku u particiji koja će biti konzumirana). Ako ofset pogrešno postavi, ista poruka može biti konzumirana više puta ili se može desiti gubitak poruka.

"Trajno čuvanje poruka:" Kafka čuva poruke u lokalnom fajl sistemu uz izuzetno visoke performanse.

"Slanje u grupama:" Kafka podržava slanje poruka organizovanih u skupove radi povećanja efikasnosti.

"Push-and-Pull:" Producer i Consumer u Kafci koriste model Push-and-Pull: Producer gura (Push) poruke ka Broker-u, a Consumer povlači (Pull) poruke iz Broker-a.

"Mehanizam particija (Partition):" Broker Kafke podržava particionisanje poruka. Producer može da odluči u koju particiju će poslati poruku. Redosled poruka unutar jedne particije odgovara redosledu kojim ih je Producer poslao. Broj particija unutar teme je konfigurabilan i Partition je važna garancija visokog protoka Kafke.

"Sistemska arhitektura"

Uobičajeno, arhitektura Kafke obuhvata "više Producer-a", "više Consumer-a", "više Broker-a" i "jedan ZooKeeper klaster".

"Producer": proizvođač; šalje poruke u Kafku.

"Consumer": potrošač; povlači poruke iz Kafke radi konzumiranja.

"Broker": Kafka čvor; jedan ili više Broker-a čine Kafka klaster.

"ZooKeeper klaster": upravlja metapodacima Kafka klastera i izborom kontrolera, itd.

Particionisanje proizvođača

"Zašto particije?"

Način organizacije poruka u Kafki zapravo je trostruka struktura: tema — particija — poruka.

Svaka poruka ispod teme čuva se samo u jednoj particiji i ne čuva se u više kopija kroz više particija.

Uloga particija jeste da pruže mogućnost balansiranja opterećenja, ili drugim rečima, glavni razlog za particionisanje podataka jeste postizanje visoke skalabilnosti (Scalability) sistema.

Različite particije mogu se smeštati na mašine različitih čvorova, a operacije čitanja i pisanja podataka izvršavaju se na nivou pojedinačne particije. Tako svaki čvor može nezavisno da obrađuje zahteve za čitanje i pisanje svoje particije, a ukupnu propusnost sistema možemo povećati dodavanjem novih čvorova.

"Koje strategije particionisanja postoje?"

"Strategija particionisanja je algoritam koji odlučuje u koju particiju će Producer poslati poruku."

Kafka nudi podrazumevanu strategiju particionisanja i istovremeno podržava definisanje prilagođenih strategija.

"Prilagođena strategija particionisanja"

Ako želite sami da definišete strategiju, potrebno je da eksplicitno podesite parametar partitioner.class na strani proizvođača.

Pri pisanju programa za proizvođača, možete napisati konkretan klasu koja implementira interfejs org.apache.kafka.clients.producer.Partitioner.

Ovaj interfejs je vrlo jednostavan i definiše samo dve metode: partition() i close(). Obično je dovoljno da implementirate samo najvažniju metodu partition.

Pogledajmo potpis ove metode:

int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster);

Ovde topic, key, keyBytes, value i valueBytes predstavljaju podatke poruke, dok cluster čuva informacije o klasteru (npr. koliko ukupno tema i Broker-a ima trenutni Kafka klaster).

Kafka vam daje toliko informacija upravo kako biste ih mogli u potpunosti iskoristiti za particionisanje poruka, odnosno izračunavanje u koju particiju će biti poslata.

Samo ako u svojoj klasi implementirate metodu partition i podesite parametar partitioner.class na puno kvalifikovano ime (Fully Qualified Name) vaše klase, program proizvođača će particionisati poruke prema vašoj logici.

"Strategija kružnog raspoređivanja"

Poznata i kao Round-robin strategija, odnosno sekvencijalno dodeljivanje.

Na primer, ako tema ima 3 particije, prva poruka se šalje u particiju 0, druga u particiju 1, treća u particiju 2, i tako dalje. Kada se pošalje četvrta poruka, kruz počinje ispočetka i ona se dodeljuje particiji 0.

To je tzv. strategija kružnog raspoređivanja. To je podrazumevana strategija particionisanja koju pruža Kafka Java Producer API.

"Strategija kružnog raspoređivanja ima izvanredne rezultate u balansiranju opterećenja — uvek garantuje maksimalno ravnomernu raspodelu poruka po svim particijama, pa je podrazumevano najrazumnija strategija i jedna je od onih koje najčešće koristimo."

"Nasumična strategija"

Poznata i kao Randomness strategija. „Nasumično” znači da poruku neurčeno smeštamo u bilo koju particiju.

Implementacija metode partition za nasumičnu strategiju je vrlo jednostavna i može se napisati u sve dve linije koda:

List partitions = cluster.partitionsForTopic(topic);
return ThreadLocalRandom.current().nextInt(partitions.size());

Prvo se izračuna ukupan broj particija teme, a zatim se nasumično vraća jedan pozitivan ceo broj manji od te vrednosti.

U suštini, nasumična strategija takođe teži ravnomernom raspoređivanju podataka po particijama, ali je u praksi nešto slabija od strategije kružnog raspoređivanja, zbog čega "ako vam je stalo do ravnomerne raspodele podataka, bolje je koristiti kružno raspoređivanje". U stvari, nasumična strategija je bila strategija korišćena u starim verzijama proizvođača; u novim verzijama je zamenjena kružnim raspoređivanjem.

"Strategija očuvanja redosleda po ključu poruke"

Kafka dozvoljava definisanje ključa poruke za svaku poruku, skraćeno Key.

Ovaj Key ima veliki značaj: može biti poslovno određen niz znakova, npr. šifra klijenta, broj odeljenja ili poslovni ID; može se koristiti i za predstavljanje metapodataka poruke.

Naročito u vreme kada Kafka nije podržavala vremenske oznake, inženjeri su u nekim scenarijima vreme kreiranja poruke direktno pakovali u Key.

Kada je poruci jednom dodeljen Key, možete garantovati da sve poruke sa istim Key-em idu u istu particiju. Pošto je obrada poruka unutar svake particije redosledna, ova strategija se naziva strategijom očuvanja redosleda po ključu poruke.

Implementacija metode partition za ovu strategiju je jednako jednostavna i zahteva samo dve linije koda:

List partitions = cluster.partitionsForTopic(topic);
return Math.abs(key.hashCode()) % partitions.size();

Pomenuta podrazumevana strategija particionisanja Kafke zapravo istovremeno implementira dve strategije: ako je naveden Key, koristi se strategija očuvanja redosleda po ključu; ako Key nije naveden, koristi se strategija kružnog raspoređivanja.

"Ostale strategije particionisanja"

Postoji još jedna prilično česta strategija, tzv. strategija particionisanja zasnovana na geografskoj lokaciji.

Naravno, ova strategija se obično primenjuje samo na velike Kafka klastere, naročito one koji obuhvataju gradove, države, pa čak i kontinente.

Možemo implementirati prilagođenu strategiju particionisanja na osnovu IP adrese na kojoj se nalazi Broker. Na primer, sledeći kôd:

List partitions = cluster.partitionsForTopic(topic);
return partitions.stream().filter(p -> isSouth(p.leader().host())).map(PartitionInfo::partition).findAny().get();

Iz svih particija možemo izdvojiti one čija se lider replika nalazi na jugu, a zatim nasumično odabrati jednu od njih za slanje poruke.

Algoritmi kompresije proizvođača

"Kako Kafka kompresuje poruke?"

Trenutno Kafka ima dve glavne kategorije formata poruka koje zajednica naziva verzijama V1 i V2.

Verzija V2 zvanično je uvedena u Kafki 0.11.0.0.

Bez obzira na verziju, hijerarhija poruka Kafke ima dva nivoa: skup poruka i poruka.

Jedan skup poruka sadrži više log stavki, a log stavka je mesto gde se poruka zaista nalazi.

Log poruka na najnižem nivou Kafke sastoji se od niza skupova poruka odnosno log stavki.

Kafka obično ne radi direktno sa pojedinačnim porukama, već uvek piše na nivou skupa poruka.

"Koja je svrha uvođenja verzije V2?"

Verzija V2 uglavnom otklanja pojedine nedostatke verzije V1, npr. zajednički deo poruka se izdvaja i čuva u spoljnom skupu poruka, tako da svaka poruka pojedinačno ne mora da čuva te podatke.

Primer: u verziji V1 svaka poruka je morala da prođe CRC proveru, ali u nekim slučajevima CRC vrednost poruke se menja.

Na primer, na strani Broker-a može se ažurirati polje vremenske oznake poruke, čime se nakon ponovnog izračunavanja ažurira i CRC vrednost; takođe, kada Broker vrši konverziju formata poruke (uglavnom zarad kompatibilnosti sa starim klijentskim programima), CRC vrednost se takođe menja.

Zbog svega ovoga, CRC provera za svaku poruku pojedinačno postaje suvišna — ne samo da troši prostor, već oduzima i CPU vreme, pa je u verziji V2 CRC provera poruke pomerena na nivo skupa poruka.

Verzija V2 donosi još jedno poboljšanje usko povezano sa kompresijom: promenjen je način čuvanja kompresovanih poruka.

U verziji V1 kompresovane poruke su čuvane tako što je više poruka kompresovano i zatim smešteno u polje tela spoljne poruke; u verziji V2 kompresuje se čitav skup poruka, što očigledno daje bolje rezultate kompresije.

"Kada se vrši kompresija?"

U Kafci kompresija može nastati na dva mesta: na strani proizvođača i na strani Broker-a.

Podešavanje parametra compression.type u programu proizvođača znači da se uključuje algoritam kompresije navedenog tipa.

Na primer, sledeći kôd pokazuje kako konstruisati Producer objekat sa uključenim GZIP-om:

Properties props = new Properties(); 
props.put("bootstrap.servers", "localhost:9092"); 
props.put("acks", "all"); 
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // uključuje GZIP kompresiju
props.put("compression.type", "gzip"); 
Producer producer = new KafkaProducer<>(props);

Ključna linija koda je props.put("compression.type", "gzip"), kojom se označava da Producer kao algoritam kompresije koristi GZIP.

Nakon pokretanja, svaki skup poruka koji ovaj Producer produkuje biće kompresovan GZIP-om, što značajno štedi mrežnu propusnost i zauzeće diska na Kafka Broker-u.

Postoje dva izuzetka u kojima Broker može ponovo da kompresuje poruke:

"Slučaj 1: Broker ima drugačiji algoritam kompresije od Producer-a."

Čim na strani Broker-a podesite različitu vrednost compression.type, budite oprezni: mogu se desiti neočekivane operacije kompresije/dekompresije, što se obično ispoljava naglim skokom zauzeća CPU-a na Broker-u.

"Slučaj 2: Na strani Broker-a se vrši konverzija formata poruka."

Konverzija formata poruka uglavnom služi da bi se omogućila kompatibilnost sa starim verzijama programa potrošača.

U produkciji je veoma čest slučaj da Kafka klaster istovremeno čuva poruke u više različitih formata.

Da bi ostao kompatibilan sa starim formatom, Broker će izvršiti konverziju novijih poruka u stariji format.

Tokom tog procesa vrši se dekompresija i ponovna kompresija poruka.

U opštem slučaju, ovakva konverzija formata znatno utiče na performanse; pored same kompresije, ona Kafka oduzima i Zero Copy osobinu.

"Kada se vrši dekompresija?"

Tamo gde postoji kompresija, postoji i dekompresija! Obično se dekompresija vrši u programu potrošača. Drugim rečima, nakon što Producer pošalje kompresovane poruke, Broker ih prima takve kakve jesu i takve ih i čuva. Kada Consumer zatraži te poruke, Broker ih šalje u originalnom obliku, a tek kada poruke stignu na stranu Consumer-a, on sam vrši dekompresiju i rekonstruiše originalne poruke.

"Osnovni tok: kompresija na strani Producer-a, očuvanje na strani Broker-a, dekompresija na strani Consumer-a."

Napomena: pored dekompresije na strani Consumer-a, Broker takođe vrši dekompresiju.

Svaki kompresovani skup poruka pri upisu na strani Broker-a podleže dekompresiji, a svrha joj je da se nad porukama izvrše razne provere valjanosti.

Moramo priznati da ova dekompresija u izvesnoj meri utiče na performanse Broker-a, posebno kada je reč o zauzeću CPU-a.

"Poređenje različitih algoritama kompresije"

Pre verzije Kafka 2.1.0, Kafka je podržavala tri algoritma kompresije: GZIP, Snappy i LZ4.

Od verzije 2.1.0, Kafka zvanično podržava algoritam Zstandard (skraćeno zstd).

To je algoritam kompresije koji je Facebook otvorio za javnost i koji pruža vrlo visok odnos kompresije.

U praksi se GZIP, Snappy, LZ4 i zstd ponašaju različito.

Za Kafku, gledano kroz propusnost: LZ4 > Snappy > zstd i GZIP; dok je u pogledu odnosa kompresije: zstd > LZ4 > GZIP > Snappy.

Što se tiče fizičkih resursa, Snappy troši najviše mrežne propusnosti, a zstd najmanje.

U pogledu zauzeća CPU-a, svi algoritmi su približno jednaki; pri kompresiji Snappy koristi nešto više CPU-a, dok pri dekompresiji GZIP može trošiti više CPU-a.

"Najbolje prakse"

❝

Kada je pravi trenutak za uključivanje kompresije?

❞

Jedan uslov za uključivanje kompresije jeste da mašina na kojoj radi program proizvođača ima dovoljno CPU resursa.

Pored dovoljno CPU resursa, ako su u vašem okruženju resursi propusnog opsega ograničeni, preporučujemo vam da uključite kompresiju.

Grupa potrošača

"Consumer Group je mehanizam potrošača koji Kafka pruža, a koji je skalabilan i otporan na greške".

Pošto je u pitanju grupa, u njoj logično može biti više potrošača odnosno instanci potrošača. One dele zajednički ID koji se naziva Group ID.

Svi potrošači unutar grupe koordiniraju kako bi konzumirali sve particije pretplaćenih tema.

❝

Svaku particiju sme konzumirati samo jedna instanca Consumer-a iz iste grupe potrošača.

❞

"Tri ključne osobine Consumer Group-a:"

1. Ispod Consumer Group-a može biti jedna ili više instanci Consumer-a; instanca može biti zaseban proces, ali i nit unutar istog procesa.
2. Group ID je niz znakova koji u okviru jednog Kafka klastera jednoznačno identifikuje Consumer Group.
3. Pojedinačnu particiju tema na koje su pretplaćene sve instance ispod Consumer Group-a sme konzumirati samo jedna instanca Consumer-a unutar grupe; istu particiju naravno može konzumirati i druga Group.

Kada se Consumer Group pretplati na više tema, ne zahteva se od svake instance unutar grupe da konzumira sve particije tema — ona će konzumirati poruke samo iz dela particija.

Consumer Group-i su međusobno nezavisne i ne utiču jedna na drugu; mogu se pretplatiti na isti skup tema bez međusobnog ometanja.

"Kafka koristi samo jedan jedini mehanizam, Consumer Group, a istovremeno ostvaruje oba velika modela tradicionalnih sistema za obradu poruka":

• Ako sve instance pripadaju istoj Group-i, realizuje se model reda poruka;
• Ako sve instance pripadaju različitim Group-ama, realizuje se model objavljivanja/pretplate.

"Koliko instanci Consumer-a treba da bude u jednoj Group-i?"

"U idealnom slučaju, broj instanci Consumer-a treba da bude jednak ukupnom broju particija tema na koje je Group-a pretplaćena."

Pretpostavimo da se Consumer Group pretplatila na 3 teme — A, B i C — sa redom 1, 2 i 3 particije. Tada je uobičajeno idealno za tu Group-u podesiti 6 instanci Consumer-a, jer se na taj način maksimalno ostvaruje visoka skalabilnost.

"Kako Kafka upravlja ofsetima za Consumer Group?"

"Ofset (Offset)"

Starije verzije Consumer Group-a čuvale su ofset u ZooKeeper-u.

Apache ZooKeeper je distribuirani okvir za koordinaciju, a Kafka u velikoj meri oslanja na njega radi raznih vidova koordinacije i upravljanja.

Čuvanje ofseta u spoljnom sistemu kao što je ZooKeeper ima najočigledniju prednost u tome što se smanjuje opterećenje čuvanja stanja na strani Kafka Broker-a.

Međutim, vremenom je otkriven problem: okviri poput ZooKeeper-a zapravo nisu pogodni za učestala ažuriranja putem upisa, a ažuriranje ofseta Consumer Group-a je operacija koja se dešava veoma često.

Ovakvi upisi visokog protoka znatno usporavaju performanse ZooKeeper klastera.

Zato je u novijim verzijama Consumer Group-a Kafka zajednica redizajnirala način upravljanja ofsetima i usvojila pristup čuvanja ofseta u internoj temi Kafke.

Ta interna tema je __consumer_offsets.

Strategije potrošača

"Prva strategija: Round-robin"

Podrazumevana, poznata i kao kružno raspoređivanje. Za istu grupu potrošača, odluka o particijama koje potrošač konzumira donosi se na osnovu kružnog raspoređivanja.

"Druga strategija: Range"

Za grupu potrošača, način konzumiranja se određuje tako što se ukupan broj particija podeli sa ukupnim brojem potrošača. Ako rezultat nije ceo broj, preostale particije se obično raspoređuju počevši od početka.

"Treća strategija: Sticky"

Uvedena u verziji 0.11.x i nešto se razlikuje od prethodne dve — predstavlja svojevrsnu nadogradnju Range strategije. Dok prve dve pri dodavanju novog potrošača u istu grupu ili izlasku starog ponovo ispočetka određuju način konzumiranja, Sticky prilikom dolaska novog potrošača ili odlaska starog ne započinje direktno novu Range raspodelu. Umesto toga zadržava postojeću strategiju konzumiranja preostalih potrošača, a particije napuštenog potrošača ravnomerno raspoređuje na preostale; isto važi i za novog potrošača, koji se odvaja od strategije konzumiranja ostalih postojećih potrošača.

Podnošenje ofseta

Pretpostavimo da u jednoj particiji ima 10 poruka sa ofsetima od 0 do 9.

Ako je neki Consumer konzumirao 5 poruka, to znači da je konzumirao poruke sa ofsetima od 0 do 4. Tada je ofset tog Consumer-a 5 i on pokazuje na ofset sledeće poruke.

Pošto Consumer može istovremeno da konzumira podatke iz više particija, podnošenje ofseta se vrši na nivou pojedinačne particije, odnosno "Consumer mora da podnese poseban ofset za svaku particiju koja mu je dodeljena".

"Podnošenje ofseta se deli na automatsko i ručno; posmatrano sa strane Consumer-a, deli se na sinhrono i asinhrono podnošenje".

Način za uključivanje automatskog podnošenja: na strani Consumer-a postoji parametar enable.auto.commit koji je dovoljno podesiti na true ili ga uopšte ne podesiti (pošto mu je podrazumevana vrednost true). Drugim rečima, Java Consumer podrazumevano automatski podnosi ofset.

Ako je automatsko podnošenje uključeno, na strani Consumer-a postoji još jedan parametar: auto.commit.interval.ms.

Njegova podrazumevana vrednost je 5 sekundi, što znači da će Kafka svakih 5 sekundi automatski podneti ofset.

Properties props = new Properties();
     props.put("bootstrap.servers", "localhost:9092");
     props.put("group.id", "test");
     props.put("enable.auto.commit", "true");
     props.put("auto.commit.interval.ms", "2000");
     props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
     KafkaConsumer consumer = new KafkaConsumer<>(props);
     consumer.subscribe(Arrays.asList("foo", "bar"));
     while (true) {
         ConsumerRecords records = consumer.poll(100);
         for (ConsumerRecord record : records)
             System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
     }

Treća i četvrta linija koda gore predstavljaju način za uključivanje automatskog podnošenja ofseta.

Način za uključivanje ručnog podnošenja jeste da se enable.auto.commit postavi na false.

Uz to je potrebno pozvati odgovarajući API za ručno podnošenje ofseta. Najsimpliji API je "KafkaConsumer#commitSync()".

Ova metoda podnosi najnoviji ofset koji je vratila metoda KafkaConsumer#poll().

Ime govori da je u pitanju sinhrona operacija — metoda će čekati dokle god ofset ne bude uspešno podnet, i tek tada se vraća.

Ako tokom podnošenja dođe do izuzetka, metoda će ga proslediti.

Sledeći kôd prikazuje način korišćenja metode commitSync():

while (true) {
            ConsumerRecords records =
                        consumer.poll(Duration.ofSeconds(1));
            process(records); // obrada poruka
            try {
                        consumer.commitSync();
            } catch (CommitFailedException e) {
                        handle(e); // obrada izuzetka neuspešnog podnošenja
            }
}

Kada jednom podesite enable.auto.commit na true, Kafka garantuje da će pri pozivu metode poll podneti sve poruke vraćene prethodnim pozivom poll.

Sa gledišta redosleda, logika metode poll je da prvo podnese ofset prethodne grupe poruka, a zatim obradi sledeću grupu poruka, pa se na taj način sprečava gubitak konzumiranih poruka.

Problem automatskog podnošenja ofseta je, međutim, u tome što "može doći do dvostrukog konzumiranja".

Ručno podnošenje ofseta je s druge strane prednost jer je fleksibilnije — vi u potpunosti kontrolišete trenutak i učestalost podnošenja ofseta.

Ipak, ono ima i jednu manu: pri pozivu commitSync() program Consumer-a biće blokiran sve dok udaljeni Broker ne vrati rezultat podnošenja.

Zbog ovog problema, Kafka zajednica za ručno podnošenje ofseta pruža još jedan API: "KafkaConsumer#commitAsync()".

Ime otkriva da nije sinhrona, već asinhrona operacija.

Nakon poziva commitAsync(), metoda se odmah vraća i ne blokira, pa ne utiče na TPS aplikacije Consumer-a.

Pošto je asinhrona, Kafka pruža funkciju povratnog poziva (callback) pomoću koje možete implementirati logiku nakon podnošenja — npr. vođenje logova ili obradu izuzetaka.

Sledeći kôd prikazuje način poziva commitAsync():

while (true) {
            ConsumerRecords records = 
 consumer.poll(Duration.ofSeconds(1));
            process(records); // obrada poruka
            consumer.commitAsync((offsets, exception) -> {
 if (exception != null)
 handle(exception);
 });
}

Problem metode commitAsync je u tome što pri pojavi problema ne radi automsko ponavljanje.

Očigledno, za ručno podnošenje potrebno je kombinovati commitSync i commitAsync kako bi se dobio optimalan rezultat. Razlozi su dva:

1. Možemo iskoristiti automatsko ponavljanje commitSync-a da izbegnemo prolazne greške, npr. trenutne mrežne treptaje, GC na strani Broker-a i sl., pošto su ovakvi problemi kratkotrajni i automatsko ponavljanje obično uspe.
2. Ne želimo da program stalno bude u blokiranom stanju, jer to utiče na TPS.

Pogledajmo sledeći kôd koji prikazuje kako kombinovati ova dva API-ja za ručno podnošenje.

   try {
           while(true) {
                        ConsumerRecords records = 
                                    consumer.poll(Duration.ofSeconds(1));
                        process(records); // obrada poruka
                        commitAysnc(); // koristi asinhrono podnošenje da izbegne blokiranje
            }
} catch(Exception e) {
            handle(e); // obrada izuzetka
} finally {
            try {
                        consumer.commitSync(); // poslednje podnošenje vrši se sinhronim blokirajućim podnošenjem
 } finally {
      consumer.close();
}
}

Razmotrimo sledeći scenario: metoda poll ne vraća 500 poruka, već 5000.

Tada sigurno nećete želeti da ofset podnesete tek nakon obrade svih 5000 poruka, jer u slučaju greške na polovini, sve do tada obrađeno mora iznova.

Na primer, u slučaju 5000 poruka možete želeti da podnesete ofset nakon svake obradjene stotine poruka, čime izbegavate ponovno konzumiranje velikih količina poruka u slučaju greške.

Kafka Consumer API za ručno podnošenje nudi sledeće metode: commitSync(Map) i commitAsync(Map).

Njihov parametar je Map objekat čiji je ključ TopicPartition, odnosno particija koja se konzumira, a vrednost je OffsetAndMetadata objekat koji uglavnom čuva podatke o ofsetu.

❝

Kako podneti ofset nakon svake obradjene stotinice poruka?

❞

Ovde ćemo, na primeru commitAsync, pokazati kôd; zapravo je način poziva commitSync potpuno isti.

private Map offsets = new HashMap<>();
int count = 0;
……
while (true) {
            ConsumerRecords records = 
 consumer.poll(Duration.ofSeconds(1));
            for (ConsumerRecord record: records) {
                        process(record); // obrada poruke
                        offsets.put(new TopicPartition(record.topic(), record.partition()),
                                   new OffsetAndMetadata(record.offset() + 1);
                       if(count % 100 == 0)
                                    consumer.commitAsync(offsets, null); // logika povratnog poziva je null
                        count++;
 }
}

Program prvo kreira Map objekat u koji se čuvaju ofseti particija koje treba podneti tokom obrade od strane Consumer-a, a zatim počinje da obrađuje poruke jedna po jedna i da gradi vrednosti ofseta koje će biti podnete.

Poslednji deo koda vrši podnošenje ofseta. Podešen je brojač koji podnosi ofset na svakih 100 obrađenih poruka.

Za razliku od poziva bez argumenata, ovde se poziva commitAsync sa argumentom Map objekta, čime se vrši podnošenje ofseta u finijoj granskoj kontroli.

Na ovaj način ovaj kôd omogućava podnošenje ofseta nakon svake stotinice obrađenih poruka, bez ograničenja na ukupan broj poruka vraćenih metodom poll.

Rebalans

"Rebalance je u suštini protokol koji određuje kako će svi Consumer-i iz jedne Consumer Group postići dogovor o raspodeli svake particije pretplaćene teme".

Na primer, neka Group ima 20 instanci Consumer-a i pretplaćena je na temu sa 100 particija.

U normalnim uslovima, Kafka će svakom Consumer-u u proseku dodeliti 5 particija. Proces dodeljivanja se naziva Rebalance.

"Rebalance se pokreće u tri slučaja."

1. Broj članova grupe se promeni. Na primer, nova instanca Consumer-a pristupi grupi ili je napusti, ili instanca Consumer-a padne i bude izbačena iz grupe.
2. Broj tema na koje je grupa pretplaćena se promeni. Consumer Group može pretplatiti se na teme putem regularnog izraza, npr. consumer.subscribe(Pattern.compile("t.*c")) znači da je Group-a pretplaćena na sve teme koje počinju slovom t, a završavaju slovom c. Ako tokom rada Consumer Group-a napravite novu temu koja zadovoljava taj uslov, ta Group će proći kroz Rebalance.
3. Broj particija teme na koju je pretplaćena se promeni. Kafka trenutno dozvoljava samo povećanje broja particija teme; kada se broj particija poveća, sve grupe pretplaćene na tu temu pokrenuće Rebalance.

Kada nastupi Rebalance, sve instance Consumer-a unutar Group-a koordiniraće zajednički.

"Strategija dodeljivanja"

Trenutno Kafka podrazumevano pruža tri strategije dodeljivanja. Svaka od njih ima određene prednosti i nedostatke, a zajednica stalno unapređuje ove strategije kako bi obezbedila najpravedniju raspodelu, pri kojoj svaka instanca Consumer-a dobije približno jednak broj particija.

Na primer, unutar jedne Group-e ima 10 instanci Consumer-a koje treba da konzumiraju 100 particija; idealna strategija dodeljivanja je da svaka instanca u proseku dobije 10 particija.

To se naziva pravednom strategijom dodeljivanja.

Pogledajmo jednostavan primer koji objašnjava proces Rebalance-a Consumer Group-a.

Pretpostavimo da Consumer Group trenutno ima dva Consumer-a, A i B. Kada se pridruži treći član C, Kafka će pokrenuti Rebalance i na osnovu podrazumevane strategije dodeljivanja ponovo raspodeliti particije između A, B i C.

Nakon Rebalance-a, raspodela je i dalje pravedna — svaka instanca Consumer-a dobija pravo konzumiranja po dve particije.

Tokom Rebalance-a, sve instance Consumer-a prestaju sa konzumiranjem i čekaju da Rebalance završi. To je jedan od najčesto kritikovanih aspekata Rebalance-a.

Trenutni dizajn Rebalance-a podrazumeva da sve instance Consumer-a učestvuju i da se sve particije ponovo raspodele.

"U kojim slučajevima će Coordinator proceniti da je neka instanca Consumer-a pala te da mora biti izbačena iz grupe?"

Kada Consumer Group završi Rebalance, svaka instanca Consumer-a periodično šalje zahtev za heartbeat Coordinatoru, čime pokazuje da je i dalje živa.

Ako neka instanca Consumer-a ne uspe blagovremeno da pošalje te heartbeat zahteve, Coordinator će proceniti da je taj Consumer mrtav, izbaciti ga iz Group-a i pokrenuti novi Rebalance.

Na strani Consumer-a postoji parametar koji se zove session.timeout.ms.

Podrazumevana vrednost ovog parametra je 10 sekundi, što znači da ako Coordinator u toku 10 sekundi ne primi heartbeat od nekog Consumer-a ispod Group-a, smatraće da je ta instanca Consumer-a pala.

Pored ovog parametra, Consumer pruža i parametar koji vam omogućava da kontrolišete učestalost slanja heartbeat zahteva — heartbeat.interval.ms.

Što je ova vrednost manja, instanca Consumer-a šalje heartbeat zahteve češće.

Često slanje heartbeat zahteva dodatno troši propusni opseg, ali je prednost to što se brže saznaje da li je Rebalance u toku, jer Coordinator trenutno obaveštava pojedinačne instance Consumer-a o pokretanju Rebalance-a tako što u telo odgovora na heartbeat zahtev ugrađuje oznaku REBALANCE_NEEDED.

Pored ova dva parametra, na strani Consumer-a postoji još jedan parametar koji kontroliše uticaj stvarne sposobnosti konzumiranja na Rebalance — max.poll.interval.ms.

Ovaj parametar ograničava maksimalni vremenski interval između dva uzastopna poziva metode poll na strani aplikacije Consumer-a.

Podrazumevana vrednost mu je 5 minuta, što znači da vaš program Consumer-a ako za 5 minuta ne uspe da konzumira sve poruke vraćene metodom poll, instanca Consumer-a će sama pokrenuti zahtev za napuštanje grupe, a Coordinator će započeti novi Rebalance.

"Konfiguracije kojima se izbegava Rebalance"

Prva kategorija Rebalance-a nastaje zbog neblagovremenog slanja heartbeat-a, zbog čega Consumer bude izbačen iz Group-a.

Zato je moguće podesiti vrednosti "session.timeout.ms i heartbeat.interval.ms".

• Podesiti session.timeout.ms = 6s.
• Podesiti heartbeat.interval.ms = 2s.
• Treba obezbediti da instanca Consumer-a pre nego što bude proglašena mrtvom pošalje bar 3 runde heartbeat zahteva, odnosno session.timeout.ms >= 3 * heartbeat.interval.ms.

Vrednost session.timeout.ms od 6s postavljena je prvenstveno kako bi Coordinator brže identifikovao Consumer-e koji su pali.

"Druga kategorija Rebalance-a nastaje jer Consumer previše vremena provodi u konzumiranju".

Trebalo bi da vašoj logici poslovne obrade ostavite dovoljno vremena, tako da Consumer ne zapadne u Rebalance zbog predugog vremena obrade poruka.

ConsumerOffsets

"Kafka čuva podatke o ofsetima Consumer-a kao obične Kafka poruke i podnosi ih u temu __consumer_offsets."

"Glavna uloga teme __consumer_offsets jeste čuvanje informacija o ofsetima Kafka potrošača."

Ovaj proces podnošenja mora ne samo da obezbedi visoku postojanost, već i da podrži učestale operacije upisa.

Tema __consumer_offsets je obična Kafka tema. Možete je ručno kreirati, menjati, pa čak i obrisati.

Iako je __consumer_offsets obična Kafka tema, "njen format poruka je samostalno definisan od strane Kafke" i korisnici ga ne mogu menjati. Drugim rečima, ne možete proizvoljno slati poruke u ovu temu, jer ako poruka koju upišete ne zadovoljava format koji je Kafka propisala, Kafka neće moći da je uspešno parsira, što može dovesti do pada Broker-a.

Kafka Consumer ima API koji vam pomaže da podnesete ofset, odnosno da upišete poruku u temu __consumer_offsets. Nikako ne pišite sopstveni Producer koji proizvoljno šalje poruke u tu temu.

__consumer_offsets ima tri formata poruka:

1. Poruka koja čuva informacije o Consumer Group-i.
2. Poruka koja briše zastarele ofsete grupe ili čak i samu grupu.
3. Poruka koja čuva vrednost ofseta.

Drugi format ima posebno ime: tombstone poruka, odnosno poruka „nadgrobni spomenik”, poznata i kao delete mark. Njena glavna osobina je da joj telo poruke jeste null, odnosno prazno telo poruke.

Kada sve instance Consumer-a ispod neke Consumer Group-e prestanu sa radom, a njihovi podaci o ofsetima budu obrisani, Kafka će upisati tombstone poruku u odgovarajuću particiju teme __consumer_offsets, čime se označava da treba temeljno obrisati podatke o toj grupi.

Kako nastaje tema __consumer_offsets?

Uobičajeno, "kada se pokrene prvi Consumer program u Kafka klasteru, Kafka automatski kreira temu za ofsete".

"Podrazumevani broj particija te teme je 50, a broj replika je 3".

Trenutno Kafka Consumer pošalje ofset na dva načina: "automatskim podnošenjem i ručnim podnošenjem."

Na strani Consumer-a postoji parametar enable.auto.commit. Ako je vrednost true, Consumer u pozadini za vas periodično podnosi ofset; interval podnošenja kontroliše se posebnim parametrom auto.commit.interval.ms.

Automatsko podnošenje ofseta ima jednu očiglednu prednost — jednostavnost: ne morate da brinete o podnošenju ofseta, a opet garantuje da konzumiranje poruka neće dovesti do gubitka.

Ujedno je to i mana: gubi se velika doza fleksibilnosti i kontrole, te uopšte ne možete da upravljate ofsetima na strani Consumer-a.

Kafka Consumer API pruža metode za podnošenje ofseta, kao što je consumer.commitSync.

Kada pozovete ove metode, Kafka upisuje odgovarajuću poruku u temu __consumer_offsets.

Ako odaberete automatsko podnošenje, može se pojaviti problem: sve dok je Consumer pokrenut, on će neograničeno upisivati poruke u temu ofsetova.

"Pogledajmo jedan ekstreman primer."

Pretpostavimo da je Consumer konzumirao najsvežiju poruku neke teme, sa ofsetom 100, i da nakon toga ta tema ne dobija nikakve nove poruke, pa Consumer nema šta da konzumira i ofset ostaje na 100.

Pošto je podnošenje automatsko, u temu ofsetova će neprestano biti upisivana poruka sa ofsetom = 100.

Očigledno je Kafkau dovoljno da od svih takvih poruka zadrži samo najsvežiju, dok se sve starije mogu obrisati.

To zahteva da Kafka poseduje strategiju brisanja poruka koja je prilagođena osobenostima teme ofsetova, jer u suprotnom ovakvih poruka bude sve više, što bi na kraju ispunilo čitav disk.

"Strategija Compact"

Kafka koristi "strategiju Compact" kako bi obrisala zastarele poruke iz teme __consumer_offsets i sprečila njen neograničeni rast.

Na primer, za isti Key postoe dve poruke, M1 i M2. Ako je vreme slanja poruke M1 ranije od vremena slanja M2, onda je M1 zastarela poruka.

Proces Compact-a sastoji se u tome da se skeniraju sve poruke iz loga, izbace zastarele, a preostale sažmu na jedno mesto.

Ovde nalazimo sliku sa zvaničnog sajta koja objašnjava proces Compact.

Na slici poruke sa ofsetima 0, 2 i 3 imaju isti Key K1. Nakon Compact-a, particija treba da čuva samo poruku sa ofsetom 3, jer je ona najsvežija.

"Kafka poseduje posebnu pozadinsku nit koja periodično obilazi teme koje treba Compact-ovati i proverava da li postoje podaci koji ispunjavaju uslove za brisanje".

Ova pozadinska nit se zove Log Cleaner.

U mnogim realnim produkcionim okruženjima se dogodilo da tema ofsetova neograničeno raste i zauzima previše disk prostora. Ako se i u vašem okruženju pojavi ovaj problem, preporučujemo vam da proverite stanje niti Log Cleaner — obično je uzrok to što je ta nit pala.

Mehanizam replika

Prema definiciji Kafka mehanizma replika, sve replike ispod jedne particije čuvaju isti niz poruka, a te replike su raspoređene na različite Broker-e, čime se suprotstavlja nedostupnosti podataka uzrokovanoj padom pojedinih Broker-a.

Slika ispod prikazuje raspored replika na Kafka klasteru sa 3 Broker-a.

Sa ove slike se vidi da se 3 replike teme 1 particije 0 nalaze na 3 različita Broker-a. I replike ostalih tematskih particija su razbacane po različitim Broker-ima, čime se ostvaruje suvišnost podataka.

"Uloge replika"

U Kafci replike se dele u dve kategorije: lider replika (Leader Replica) i pratilec replika (Follower Replica).

Pri kreiranju svake particije bira se jedna replika koja se naziva lider replika, dok se ostale automatski nazivaju pratilec replikama.

U Kafci pratilec replike ne pružaju usluge ka spolja. Drugim rečima, nijedna pratilec replika ne može da odgovori na zahteve za čitanje i pisanje potrošača i proizvođača. Sve zahteve mora obraditi lider replika, odnosno svi zahtevi za čitanje i pisanje moraju biti poslati Broker-u na kom se nalazi lider replika, koji ih i obrađuje.

Pratilec replika ne obrađuje zahteve klijenata; njen jedini zadatak jeste da "asinhrono povuče" poruke od lider replike i upiše ih u sopstveni log predaje, čime se ostvaruje sinhronizacija sa lider replikom.

Kada lider replika padne, ili kada Broker na kom se nalazi lider replika otkaže, Kafka se oslanja na mogućnost nadgledanja koju pruža ZooKeeper i u realnom vremenu to detektuje, te odmah pokreće novi izbor lidera — iz redova pratilec replika bira jednu kao novog lidera. Nakon što stari Leader replika ponovo krene, može se vratiti u klaster samo kao pratilec replika.

Za klijente korisnike, Kafka pratilec replike nemaju nikakvu ulogu. Zašto ih onda Kafka uopšte ima?

Ovaj mehanizam replika donosi dve vrste koristi.

1. "Olakšava ostvarivanje Read-your-writes".

Read-your-writes, kako i samo ime kaže, znači da kada pomoću API-ja proizvođača uspešno upišete poruku u Kafku, istog trenutka pomoću API-ja potrošača možete pročitati tu poruku.

2. "Olakšava ostvarivanje monotong čitanja (Monotonic Reads)".

Pretpostavimo da trenutno postoje dve pratilec replike, F1 i F2, koje asinhrono povlače podatke od lider replike. Ako je F1 povukla najnovije poruke od Leader-a, a F2 to još nije stigla, tada može se desiti sledeća situacija: potrošač koji prvo čita iz F1, a zatim povlači iz F2, može videti da je najnovija poruka koju je video pri prvom konzumiranju nestala u drugom konzumiranju — i to nije konzistentnost monotong čitanja.

Međutim, ako sve zahteve za čitanje obrađuje Leader, Kafki je lako da ostvari konzistentnost monotong čitanja.

ISR mehanizam

In-sync Replicas, odnosno skup ISR replika.

Replake koje se nalaze u ISR-u su one koje su u sinhronizaciji sa Leaderom. Suprotno tome, pratilec replike koje nisu u ISR-u smatraju se nesinhronizovanim sa Leaderom.

❝

Koje replike mogu ući u ISR?

❞

Lider replika je prirodno u ISR-u. Drugim rečima, "ISR nije samo skup pratilec replika; on nužno obuhvata i Leader repliku. Postoje čak i situacije u kojima ISR ima samo tu jednu Leader repliku".

Pored toga, pratilec replika koja želi ući u ISR mora ispunjavati određeni uslov.

"Preko vrednosti parametra replica.lag.time.max.ms na strani Broker-a".

Ovaj parametar označava najduže vremensko razdoblje za koje pratilec replika sme da zaostaje za lider replikom; trenutna podrazumevana vrednost mu je 10 sekundi.

To znači da, sve dok jedna pratilec replika neprekidno zaostaje za Leaderom manje od 10 sekundi, Kafka smatra da je ta pratilec replika u sinhronizaciji sa Leaderom, čak i ako u tom trenutku u pratilec replici ima znatno manje poruka nego u lider repliki.

Jedini posao pratilec replike jeste da neprestano povlači poruke od Leadera i upisuje ih u sopstveni log predaje.

Ako ta replika vremenom stigne do napretka Leadera, može ponovo biti vraćena u ISR.

ISR je dinamički promenljiv skup, a ne statičan.

Unclean izbor lidera

"Kafka sve žive replike koje nisu u ISR-u naziva nesinhronizovanim replikama".

Obično su nesinhronizovane replike znatno iza Leadera, tako da ako se neka od njih izabere za novog Leadera, može doći do gubitka podataka.

Tokom onih replika se čuvaju poruke znatno starije od onih u starom Leaderu.

U Kafci se proces izbora takvih replika naziva Uncain izbor lidera.

"Parametar na strani Broker-a unclean.leader.election.enable kontroliše da li je dozvoljen Unclean izbor lidera".

Uključivanje Unclean izbora lidera može izazvati gubitak podataka, ali je korist u tome što lider replika particije uvek postoji, pa usluge ka spolja ne moraju da prekinu, što povećava visoku dostupnost. Obrnuto, zabrana Unclean izbora lidera ima korist u tome što održava konzistentnost podataka i izbegava gubitak poruka, ali žrtvuje visoku dostupnost.

Izbor replika

Za postavljanje particija i lidera replika proizvoljne teme unutar Kafka klastera potrebno je uzeti u obzir uravnoteženost ukupne sposobnosti opterećenja klastera. Trudiće se da lider replike različitih particija budu raspoređene na različite Broker-e, kako bi se izbeglo da više lidera bude na istom Brokeru, što bi dovelo do neuravnoteženosti opterećenja među Brokerima u klasteru.

Kafka uvodi pojam preferirane replike. Preferirana replika je prva replika u listi AR skupa (sve replike particije); u idealnom stanju ta replika jeste lider replika te particije.

Na primer, Kafka klaster čine 3 Broker-a. Napravljena je tema pod imenom topic-partitions, sa 3 particije i 3 replike. Za particiju 0 AR lista je [1,2,0], što znači da je preferirana replika particije 0 upravo Broker 1.

Kafka mehanizmom više replika povećava pouzdanost, ali samo lider replika pruža usluge čitanja i pisanja ka spolja, dok pratilec replike samo sinhronizuju poruke.

"Ako lider replika neke particije postane nedostupna, to znači da cela particija postaje nedostupna, pa je tada potrebno iz redova pratilec replika izabrati novog lidera koji će nastaviti da pruža usluge".

"Pri kreiranju teme, particije te teme i replike će biti što ravnomernije raspoređene po različitim Brokerima Kafka klastera".

Na primer, kada na Kafka klasteru sa 3 Broker čvora kreiramo temu topic-partitions sa 3 particije i faktorom replikacije 3, lider replike će biti ravnomerno raspoređene na sva 3 Broker čvora.

"Za istu particiju nije moguće da se na istom Broker čvoru nalazi više njenih replika".

Čvor na kom se nalazi lider replika može se nazvati lider čvorom te particije, a čvor na kom se nalazi pratilec replika može se nazvati pratilec čvorom.

U gornjem primeru, lider čvor particije 0 je broker1, lider čvor particije 1 je broker2, a lider čvor particije 2 je broker0.

Kada lider čvor neke particije otkazuje, jedan od pratilec čvorova bira se za novog lider čvora.

Kada se originalni lider čvor oporavi, on može postati samo pratilec čvor, što dovodi do neuravnoteženosti opterećenja klastera.

Na primer, lider čvor particije 1, broker2, pao je. U tom trenutku je pratilec čvor particije 1 na broker1 izabran za novog lider čvora.

Kada broker2 ponovo postane dostupan, stanje Kafka klastera je sledeće:

Vidi se da je sada opterećenje na broker1 veće, dok broker2 nema opterećenja.

"Da bi se rešio gorenavedeni problem neuravnoteženosti opterećenja, Kafka podržava izbor preferirane replike; preferirana replika je prva replika u AR skupu particije".

Na primer, za gorenavedenu particiju 1, njen AR skup je [2,0,1], što znači da je preferirana replika particije 1 na broker2.

U idealnom slučaju, preferirana replika bi trebalo da bude lider replika. Kafka garantuje ravnomernu raspodelu preferiranih replika, što ne zavisi od toga da li je neki Broker čvor pao ili ne.

"Izbor preferirane replike znači da pri izboru lider replike particije treba, koliko god je moguće, preferiranu repliku postaviti za lider repliku". Za gorenavedenu situaciju dovoljno je ponovo pokrenuti izbor preferirane replike i tako osigurati ravnotežu opterećenja između particija.

Kafka podržava automatski izbor preferirane replike — podrazumevano se svakih 5 minuta pokreće operacija izbora preferirane replike.

Model mrežne komunikacije

U Broker-u postoji Acceptor (mainReactor) koji osluškuje dolazak novih veza. Nakon uspostavljanja veze sa novim klijentom, kružnim redosledom bira se Processor (subReactor) koji upravlja tom vezom.

Processor osluškuje događaje na vezama kojima upravlja. Kada događaj stigne, čita ga, pakuje u Request i stavlja taj Request u deljeni red zahteva.

Zatim bazen IO niti neprestano uzima zahteve iz tog reda i izvršava stvarnu obradu. Nakon obrade, odgovor se šalje u red odgovora odgovarajućeg Processor-a, a zatim Processor vraća Response klijentu.

Svaki listener ima samo jednu Acceptor nit, jer ona služi samo za uspostavljanje veze i njenu distribuciju, bez suvišne logike — vrlo je laka.

Processor se u Kafci naziva mrežnom niti. Podrazumevano, bazen mrežnih niti ima 3 niti, a odgovarajući parametar je num.network.threads i može se dinamički povećavati ili smanjivati u skladu sa stvarnim potrebama biznisa.

Postoji i bazen IO niti, odnosno KafkaRequestHandlerPool, koji izvršava stvarnu obradu. Odgovarajući parametar je num.io.threads, a podrazumevana vrednost mu je 8.

Kada IO niti završe obradu, stavljaju Response u odgovarajući Processor, a Processor vraća odgovor klijentu.

Vidi se da između mrežnih i IO niti deluje klasičan obrazac proizvođač — potrošač, bilo da je reč o deljenom redu zahteva za obradu Request-a ili o Response-u koji IO obrada vraća.

Idempotencija

"Idempotentni Producer"

U Kafci Producer podrazumevano nije idempotentan, ali možemo kreirati idempotentnog Producera.

To je zapravo nova funkcija uvedena u verziji 0.11.0.0. Pre nje, kada je Kafka slala podatke u particiju, moglo se desiti da ista poruka bude poslata više puta, što je dovelo do duplikata poruka.

Od verzije 0.11 nadalje, način za postavljanje idempotencije Producera je vrlo jednostavan — dovoljno je podesiti samo jedan parametar, i to:

props.put("enable.idempotence", true),
ili props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true).

Kada se enable.idempotence postavi na true, Producer se automatski unapređuje u idempotentnog Producera, bez potrebe za bilo kakvom izmenom preostalog koda.

Kafka automatski vrši deduplikaciju poruka za vas.

Konkretan princip na nižem nivou je vrlo jednostavan: klasičan optimizacioni pristup razmene prostora za vreme — na strani Broker-a čuva se nekoliko dodatnih polja.

Kada Producer pošalje poruke sa istim vrednostima tih polja, Broker automatski shvata da su te poruke duplikati i može ih u pozadini neprimećeno odbaciti.

"Domet idempotentnog Producera"

Pre svega, on može da garantuje idempotenciju samo na nivou jedne particije, odnosno idempotentni Producer može da garantuje da se na jednoj particiji neke teme neće pojaviti duplikati poruka, ali ne može ostvariti idempotenciju na više particija.

Drugo, on može ostvariti idempotenciju samo unutar jedne sesije, a ne i kroz više sesija.

Ovde pod sesijom možete shvatiti jedno pokretanje procesa Producera; kada restartujete proces Producera, ova garancija idempotencije se gubi.

Transakcije

Kafka od verzije 0.11 takođe pruža podršku za transakcije, i to trenutno pretežno na nivou izolacije read committed.

To omogućava da se više poruka atomski upiše u ciljne particije, a istovremeno garantuje da Consumer može videti samo poruke uspešno potvrđene transakcije.

"Transakcioni Producer"

Transakcioni Producer može da garantuje atomsko upisivanje poruka u više particija.

Te poruke ili bude sve uspešno upisane, ili sve budu neuspešne. Pored toga, transakcioni Producer se ne plaši restartovanja procesa.

Kada se Producer ponovo pokrene, Kafka i dalje garantuje obradu poruka tih Producera tačno jednom (exactly-once).

Način postavljanja transakcionog Producera je takođe jednostavan i dovoljno je ispuniti dva uslova:

• Kao i kod idempotentnog Producera, uključiti enable.idempotence = true.
• Podesiti parametar na strani Producera transactional.id; najbolje mu je dati smisleno ime.

Pored toga, potrebno je izvršiti i nekoliko izmena u kodu Producera, kao što sledi:

producer.initTransactions();
try {
            producer.beginTransaction();
            producer.send(record1);
            producer.send(record2);
            producer.commitTransaction();
} catch (KafkaException e) {
            producer.abortTransaction();
}

U poređenju sa kodom običnog Producera, upadljiva osobina transakcionog Producera jeste u tome što poziva određene API-je za transakcije: initTransaction, beginTransaction, commitTransaction i abortTransaction, koji redom odgovaraju inicijalizaciji transakcije, početku transakcije, potvrdi transakcije i prekidu transakcije.

Ovaj kôd garantuje da se Record1 i Record2 kao jedna transakcija zajednički podnesu Kafki — ili budu uspešno potvrđene, ili pak neuspešno upisane.

U stvari, čak i ako je upis neuspešan, Kafka će ih ipak upisati u log na nižem nivou, što znači da će ih Consumer i dalje videti.

Postoji parametar isolation.level koji može imati dve vrednosti:

1. read_uncommitted: ovo je podrazumevana vrednost i znači da Consumer može pročitati bilo koju poruku koju Kafka upiše, bez obzira na to da li transakcioni Producer potvrđuje ili prekida transakciju; ako koristite transakcioni Producer, odgovarajući Consumer ne bi trebalo da koristi ovu vrednost.
2. read_committed: znači da će Consumer čitati samo poruke koje je transakcioni Candidate uspešno potvrdio, a može videti i sve poruke koje upiše netransakcioni Producer.

Interceptori

"Kafka interceptor se deli na interceptor proizvođača i interceptor potrošača".

Interceptor proizvođača omogućava da svoju logiku interceptor-a ugradite pre slanja poruke i nakon uspešnog podnošenja poruke;

Interceptor potrošača podržava pisanje specifične logike pre konzumiranja poruke i nakon podnošenja ofseta.

Niz inter ceptora može se povezati u jedan veliki interceptor, a Kafka će redom kojim su dodati, jedan po jedan, izvršavati logiku svakog od njih.

Trenutno se podešavanje Kafka inter ceptora vrši preko parametara, a na stranama proizvođača i potrošača postoji zajednički parametar interceptor.classes. On navodi listu klasa, pri čemu je svaka klasa klasa implementacije interceptor-a sa specifičnom logikom.

Properties props = new Properties(); 
List interceptors = new ArrayList<>(); 
interceptors.add("com.yourcompany.kafkaproject.interceptors.AddTimestampInterceptor"); // interceptor 1
interceptors.add("com.yourcompany.kafkaproject.interceptors.UpdateCounterInterceptor"); // interceptor 2
props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors); 
……

❝

Kako napisati klase AddTimeStampInterceptor i UpdateCounterInterceptor?

❞

Ove dve klase, kao i sve klase implementacije Producer inter ceptora koje sami napišete, moraju nasleđivati interfejs org.apache.kafka.clients.producer.ProducerInterceptor.

Ovaj interfejs, koji Kafka pruža, ima dve ključne metode.

1. onSend: ova metoda se poziva pre slanja poruke.
2. onAcknowledgement: ova metoda se poziva nakon što je poruka uspešno potvrđena ili nakon što slanje ne uspe. Poziv onAcknowledgement se dešava pre poziva callback-a. Važno je napomenuti da se ova metoda i onSend ne pozivaju iz iste niti, tako da ako u ove dve metode pristupate nekom zajedničkom mutable objektu, obavezno je obezbediti thread-safety.

Isti princip važi i za podešavanje inter ceptora potrošača — jedina razlika je u tome što konkretne klase implementacije moraju da implementiraju interfejs org.apache.kafka.clients.consumer.ConsumerInterceptor. I u njemu postoje dve ključne metode.

1. onConsume: ova metoda se poziva pre nego što se poruke vrate programu Consumer-a.
2. onCommit: Consumer poziva ovu metodu nakon podnošenja ofseta. U njoj obično možete obavljati knjigovodstvene radnje, poput vođenja logova.

Važno je napomenuti da "pri navođenju klasa inter ceptora treba navesti njihova puno kvalifikovana imena".

Drugim rečima, potrebno je navesti i puno ime paketa, a ne samo ime klase. Takođe, treba obezbediti da vaš program Producera može ispravno da učita vaše klase interceptor-a.

Kontroler

"Komponenta kontrolera (Controller); njena glavna uloga je da, uz pomoć Apache ZooKeeper-a, upravlja i koordinira celokupnim Kafka klasterom".

Bilo koji Broker u klasteru može da preuzme ulogu kontrolora. Međutim, tokom rada, samo jedan Broker može biti kontrolor i vršiti dužnosti upravljanja i koordinacije.

Kafka kontroler u velikoj meri koristi Watch mogućnost ZooKeeper-a radi koordinacije i upravljanja klasterom.

"Kako se bira kontrolor"

U stvari, pri pokretanju, Broker pokušava da u ZooKeeper-u kreira čvor /controller.

Trenutno pravilo za izbor kontrolora jeste: "prvi Broker koji uspešno kreira čvor /controller biva određen za kontrolora".

"Šta radi kontrolor"

Dužnosti kontrolora se grubo mogu podeliti u 5 kategorija:

1."Upravljanje temama (kreiranje, brisanje, dodavanje particija)"

Kontrolor nam pomaže da obavimo kreiranje, brisanje i dodavanje particija Kafka tema.

2.«Reodređivanje particija»

3.«Preferred izbor lidera»

Preferred izbor lidera jeste rešenje za promenu Leader-a koje Kafka pruža kako bi izbegla preopterećenost pojedinih Broker-a.

4.«Upravljanje članovima klastera (dodavanje Broker-a, ručno gašenje Broker-a, pad Broker-a)»

Obuhvata automatsko detektovanje novih Broker-a, ručno gašenje Broker-a i detekciju pada Broker-a.

Ovo automatsko detektovanje se oslanja na kombinaciju Watch mogućnosti i ZooKeeper privremenih čvorova.

Na primer, komponenta kontrolora koristi "Watch mehanizam" da proveri promene u broju podčvorova ispod ZooKeeper čvora /brokers/ids.

Trenutno, kada se novi Broker pokrene, on ispod /brokers kreira svoj specifični znode.

Čim to bude urađeno, ZooKeeper preko Watch mehanizma šalje obaveštenje kontroloru, koji na taj način automatski oseća tu promenu i započinje dalji rad na dodavanju novog Broker-a.

Detektovanje živosti Broker-a se oslanja na gore pomenuti drugi mehanizam: «privremeni čvor».

Nakon što se Broker pokrene, on ispod /brokers/ids kreira privremeni znode.

Kada Broker padne ili bude ručno ugašen, sesija između tog Broker-a i ZooKeeper-a se završava, a taj znode se automatski briše.

Isto tako, Watch mehanizam ZooKeeper-a prosleđuje ovu promenu kontroloru, koji na taj način saznaje da je Broker ugašen ili pao, te može sprovesti naknadne radnje.

5.«Servis podataka»

Kontrolor čuva najpotpunije metapodatke o klasteru. Svi ostali Broker-i periodično primaju zahteve za ažuriranje metapodataka koje šalje kontrolor, a zatim ažuriraju podatke u svojoj memoriji.

«Prenos greške kontrolora (Failover)»

«Prenos greške (Failover) znači da kada kontrolor koji radi iznenada padne ili bude neočekivano prekinut, Kafka to brzo detektuje i odmah aktivira rezervnog kontrolora koji zamenjuje onog koji je otkazao.» Ovaj proces se naziva Failover; on se odvija automatski i ne zahteva ručno uplitanje.

U početku je Broker 0 kontrolor. Kada Broker 0 padne, ZooKeeper preko Watch mehanizma to detektuje i briše privremeni čvor /controller.

Nakon toga, svi živi Broker-i počinju da se takmiče za ulogu novog kontrolora.

Broker 3 na kraju dobija izbore i uspešno ponovo kreira čvor /controller u ZooKeeper-u.

Zatim Broker 3 iz ZooKeeper-a čita metapodatke klastera i inicijalizuje ih u sopstveni keš.

Time je Failover kontrolora završen i on može nastaviti sa normalnim radnim dužnostima.

Skladište logova

Poruke u Kafci se grupišu prema temama kao osnovnoj jedinici, a svaka tema je logički nezavisna.

Svaka tema se može podeliti na jednu ili više particija. Bez uzimanja u obzir replika, jedna particija odgovara jednom logu.

Pošto se log fajl sa vremenom neprestano širi, dizajneri su, kako bi sprečili preveliki Log, uveli pojam segmentacije loga (LogSegment) i podelili Log u više LogSegment-a, čime se olakšava naknadno održavanje i čišćenje poruka.

Slika ispod prikazuje odnos između teme, particije, replike, Log-a i LogSegment-a.

"LogSegment"

U Kafci, svaki Log objekat se može podeliti u više LogSegment fajlova. Svaki LogSegment fajl obuhvata jedan fajl podataka loga i dva indeksna fajla (indeksni fajl ofseta i indeksni fajl vremenskih oznaka poruka).

Veličina fajla podataka loga u svakom LogSegment-u je jednaka (ta veličina se može podesiti u konfiguracionom fajlu Kafka Broker-a config/server.properties preko parametra "log.segment.bytes", koja podrazumevano iznosi 1G, odnosno 1073741824 bajtova). Pri sekvencijalnom upisivanju poruka, ako se premaši ovaj prag, kreira se novi skup fajlova sa podacima loga i indeksima.

Najčešće korišćeni parametri

"Konfiguracija na strani Broker-a"

• broker.id

Svaki Kafka Broker ima jedinstveni identifikator, a taj jedinstveni identifikator jeste broker.id, čija je podrazumevana vrednost 0.

Ova vrednost mora biti jedinstvena u okviru Kafka klastera i može se proizvoljno podesiti.

• port

Ako Kafku pokrećete pomoću konfiguracionog primera, ona će osluškivati na portu 9092. Izmena parametra port omogućava da je podesite na bilo koji port.

Imajte na umu da za portove ispod 1024 treba koristiti root privilegije pri pokretanju Kafke.

• zookeeper.connect

ZooKeeper adresa koja čuva metapodatke Broker-a navodi se preko zookeeper.connect.

Na primer, navođenjem localhost:2181 kazete da taj ZooKeeper radi na lokalnom portu 2181.

Takođe možemo navesti više vrednosti za zookeeper.connect, npr. preko zk1:2181,zk2:2181,zk3:2181.

Ovaj konfiguracioni parametar predstavlja listu hostname:port/path vrednosti razdvojenih zarezima, sa sledećim značenjem:

• hostname je ime mašine ili IP adresa ZooKeeper servera.
• port je broj porta ZooKeeper klijenta.
• /path je opciona ZooKeeper putanja; Kafka ovde koristi chroot okruženje, a ako se ne navede, podrazumevano se koristi korena putanja.

❝

Ako imate dva Kafka klastera — recimo, kafka1 i kafka2 — parametar zookeeper.connect oba klastera može se navesti ovako: zk1:2181,zk2:2181,zk3:2181/kafka1 i zk1:2181,zk2:2181,zk3:2181/kafka2

❞

• log.dirs

Kafka sve poruke čuva na disk. Direktorijum u kom se čuvaju ti segmenti loga navodi se preko log.dirs; to je lista putanja lokalnog fajl sistema razdvojenih zarezima. log.dirs nema podrazumevanu vrednost, te "morate ručno podesiti njegovu vrednost".

Postoji i parametar log.dir (bez s). Podrazumevano je dovoljno da podesite samo log.dirs; na primer, vrednost ovog parametra možete podesiti kao /home/kafka1,/home/kafka2,/home/kafka3.

• auto.create.topics.enable

Podrazumevano Kafka automatski kreira teme.

Preporuka je da parametar auto.create.topics.enable podesite na false, odnosno da onemogućite automatsko kreiranje teme.

"Konfiguracije vezane za teme"

• num.partitions

Parametar num.partitions određuje broj particija koje nova tema treba da ima; podrazumevana vrednost mu je 1.

• default.replication.factor

Ovaj parametar je prilično jednostavan; on označava broj replika u koje Kafka čuva poruke.

• log.retention.ms

Kafka obično na osnovu vremena odlučuje koliko dugo će podaci biti zadržani.

Podrazumevano se za podešavanje vremena koristi parametar log.retention.hours, čija je podrazumevana vrednost 168 sati, odnosno sedam dana.

Pored njega postoje još dva parametra: log.retention.minutes i log.retention.ms.

Sva tri parametra imaju istu ulogu — određuju nakon koliko vremena će poruke biti obrisane; preporučujemo korišćenje parametra log.retention.ms.

• message.max.bytes

Broker preko parametra message.max.bytes ograničava veličinu pojedinačne poruke. Podrazumevana vrednost mu je 1000000, odnosno 1 MB. Ako pokušate da pošaljete poruku koja prelazi tu veličinu, poruka neće biti primljena, a dobićete i poruku o grešci koju vraća Broker.

• retention.ms

Određuje koliko dugo se čuvaju poruke neke teme. Podrazumevana vrednost mu je 7 dana, što znači da ta tema može čuvati poruke samo 7 dana. Kada jednom podesite ovu vrednost, ona će prepisati globalnu vrednost parametra na strani Broker-a.

Problem gubitka poruka

"Gubitak podataka u programu proizvođača"

Trenutno Kafka Producer šalje poruke asinhrono, što znači da ako pozovete API producer.send(msg), on će se obično odmah vratiti, ali u tom trenutku ne možete smatrati da je slanje poruke uspešno završeno.

Koji sve faktori, pri ovom načinu slanja, mogu uticati na to da poruka ne bude uspešno poslata?

Razloga zapravo ima mnogo — na primer, mrežni treptaj zbog kojeg poruka uopšte ne stigne do Broker-a; ili je sama poruka neispravna, pa je Broker odbija da primi (npr. poruka je prevelika i prevazilazi mogućnosti Broker-a), itd.

Rešenje ovog problema je zapravo vrlo jednostavno: Producer uvek treba da koristi API za slanje sa obaveštenjem preko callback-a. Drugim rečima, ne koristite producer.send(msg), već producer.send(msg, callback).

On vam tačno saopštava da li je poruka zaista uspešno podnesena.

Čim se pojavi situacija da poruka nije uspešno podnesena, možete preduzeti odgovarajuće postupke.

"Gubitak podataka u programu potrošača"

Gubitak podataka na strani Consumer-a pre svega se ispoljava tako da poruke koje Consumer treba da konzumira nestanu.

Sledeća slika jasno prikazuje podatke o ofsetu na strani Consumer-a.

Na primer, za Consumer A njegova trenutna vrednost ofseta jeste 9; za Consumer B vrednost ofseta je 11.

Program Consumer-a nakon preuzimanja poruka iz Kafke otvara više niti koje asinhrono obrađuju poruke, dok program Consumer-a automatski pomiče ofset unapred.

Ako jedna od tih niti ne uspe, poruke koje su njoj dodeljene neće biti uspešno obrađene, ali je ofset već ažuriran, pa te poruke za Consumer-a zapravo bivaju izgubljene.

Ključ je u tome što Consumer automatski podnosi ofset.

Rešenje ovog problema je takođe jednostavno:

"Ako se poruke obrađuju kroz višenitno asinhrono konzumiranje, program Consumer-a ne treba da uključuje automatsko podnošenje ofseta, već aplikacija treba ručno da podnosi ofset".

Najbolje prakse

Pregled konfiguracija za Kafku bez gubitka poruka:

1. Ne koristite producer.send(msg), već producer.send(msg, callback) — obavezno koristite send metodu koja podržava obaveštenje preko callback-a.
2. Podesite acks = all. acks je parametar Producera koji predstavlja vašu definiciju „potvrđene poruke”. Ako se podesi na all, to znači da poruka tek onda jeste potvrđena kada je prime sve replike Broker-a.
3. Podesite retries na veću vrednost. I ovaj retries je parametar Producera; on odgovara pomenutom automatskom ponavljanju Producera. Pri mrežnim trenutnim treptajima slanje poruke može neuspeti, a tada Producer sa retries > 0 automatski ponavlja slanje i tako izbegava gubitak poruka.
4. Podesite unclean.leader.election.enable = false. Ovo je parametar na strani Broker-a i kontroliše koji Broker-i imaju pravo da budu izabrani za lidera particije. Ako neki Broker znatno zaostaje za prethodnim Leaderom, čim postane novi Leader sigurno će dovesti do gubitka poruka, pa se ovaj parametar obično postavlja na false, čime se takva situacija sprečava.
5. Podesite replication.factor >= 3. I ovo je parametar na strani Broker-a; čuvanjem poruka u više kopija povećava se pouzdanost, a glavni mehanizam za sprečavanje gubitka poruka trenutno jeste suvišnost.
6. Podesite min.insync.replicas > 1. I ovo je parametar na strani Broker-a; on kontroliše u koliko replika minimalno poruka mora biti upisana da bi se smatrala potvrđenom. Vrednost veća od 1 povećava postojanost poruka; u stvarnim okruženjima nikako ne koristite podrazumevanu vrednost 1.
7. Obezbedite da replication.factor > min.insync.replicas. Ako su te dve vrednosti jednake, pad bilo koje replike znači da cela particija prestaje sa normalnim radom. Cilj nam je ne samo da poboljšamo postojanost poruka i sprečimo gubitak podataka, već i da to ostvarimo bez smanjenja dostupnosti; preporučena postavka je replication.factor = min.insync.replicas + 1.
8. Obezbedite da se poruka konzumira pre podnošenja ofseta. Na strani Consumer-a postoji parametar enable.auto.commit; najbolje ga je podesiti na false i koristiti ručno podnošenje ofseta.

Problem dvostrukog konzumiranja

"Scenariji u kojima se javlja dvostruko konzumiranje"

Uz podrazumevanu vrednost enable.auto.commit = true, scenariji u kojima se javlja dvostruko konzumiranje su sledeći:

❝

Tokom konzumiranja, aplikacija Consumer-a bude nasilno ubijena (kill) ili završi rad zbog izuzetka.

❞

Na primer, posle jednog poll-a 500 poruka, pri konzumiranju 200. poruke proces bude nasilno ubijen i ofset ne bude podnesen, ili pak zbog izuzetka završi rad i ofset ne bude podnesen.

Pri sledećem pokretanju ponovo će se povući 500 poruka, što dovodi do toga da prethodnih 200 konzumiranih poruka budu konzumirane još dva puta.

Rešenje: pri pojavi izuzetka pravilno obraditi nepodneseni ofset.

"Potrošač previše vremena provodi u konzumiranju»

Parametar max.poll.interval.ms definiše maksimalni razmak između dva poll-a. Njegova podrazumevana vrednost je 5 minuta, što znači da vaš program Consumer-a ako za 5 minuta ne uspe da konzumira poruke vraćene metodom poll, Consumer će samoinicijativno poslati zahtev za napuštanje grupe, a Coordinator će započeti novi Rebalance.

Primer: jedno povlačenje donese 11 poruka; obrada svake poruke traje 30 s, pa obrada 11 poruka traje 5 minuta i 30 sekundi. Pošto je max.poll.interval.ms podrazumevano 5 minuta, Consumer ne stiže da konzumira sve u roku od 5 minuta, pa napušta grupu, što dovodi do rebalance-a.

Nakon konzumiranja svih 11 poruka, Consumer će ponovo uspostaviti vezu sa Brokerom i ponovo proći kroz rebalance. Pošto ofset iz prethodnog konzumiranja nije bio podnesen, poruke koje se ponovo povuku su iste one koje su već konzumirane, što dovodi do dvostrukog konzumiranja.

"Rešenje:"

1. Povećati sposobnost konzumiranja i ubrzati obradu pojedinačne poruke; u skladu sa stvarnim scenarijem, može se povećati vrednost max.poll.interval.ms kako bi se izbegli nepotrebni rebalance-i; može se smanjiti vrednost max.poll.records, čija podrazumevana vrednost iznosi 500, u skladu sa stvarnim tempom poruka.

2. Pri slanju poruka može se uključiti jedinstveni identifikator, npr. message id. Na strani potrošača sačuvati najnovijih 1000 message id-ova u Redis ili MySQL i izvršiti deduplikaciju pre same obrade poruke.

Problem redosleda poruka

Svi znamo da topic Kafke nije uređen, ali jedan topic sadrži više partition-a, pri čemu je svaki partition interno uređen.

"Scenarij nereda 1"

Pošto jedna tema može imati više particija, Kafka može garantovati redosled samo unutar pojedinačne particije.

"Rešenje"

1. Tema se može podesiti tako da ima samo jednu particiju.

2. Prema poslovnim potrebama, poruke kojima je potreban redosled mogu se usmeriti u istu particiju.

3. Prema poslovnim potrebama, za istu narudžbinu može se koristiti isti key, što garantuje da će poruke biti raspoređene u istu particiju.

"Scenarij nereda 2"

Kada isti biznis uđe u istu grupu potrošača, a zatim se poruke obrađuju pomoću više niti, redosled poruka se narušava.

"Rešenje"

Potrošač interno, na osnovu broja niti, kreira odgovarajući broj memorijskih redova. Za niz poslovnih podataka kojima je potreban redosled, na osnovu key-a ili samih podataka, smeštaju se u jedan isti memorijski red, a zatim nit vadi podatke iz njemu odgovarajućeg memorijskog reda i nad njima vrši operacije.

"Zadržavanje redosleda poruka postavljanjem istog key-a ima jedan mali nedostatak:"

Na primer, slanje poruka je podešeno tako da ima mehanizam ponavljanja i asinhrono je. Poruke A i B imaju isti key, poslovno A ide prva, pa B. Zbog mreže ili nekog drugog razloga slanje A ne uspe, a slanje B uspe; A se, pošto nije uspelo, ponavlja i na kraju uspe. Tada je redosled poruka B pre A, što je u suprotnosti sa redosledom slanja iz biznisa. Da biste rešili ovaj problem, potrebno je podesiti parametar max.in.flight.requests.per.connection=1, koji ograničava broj neodgovorenih zahteva koje klijent sme da šalje preko jedne veze. Vrednost 1 znači da Kafka Broker dok ne odgovori na zahtev, klijent ne sme slati nove zahteve istom Brokeru. Podrazumevana vrednost ovog parametra je 5.

❝

Zvanična dokumentacija kaže da, ako je ovaj parametar veći od 1, redosled poruka se može promešati usled ponavljanja.

❞

Razlozi za visoke performanse

"Sekvencijalno čitanje i pisanje"

Poruke Kafke se neprestano nadovezuju na fajl, što omogućava kafka-i da u potpunosti iskoristi performanse sekvencijalnog čitanja i pisanja diska.

Sekvencijalno čitanje i pisanje ne zahteva vreme pomeranja glave diska (seek), već samo vrlo malo vremena rotacije sektora, pa je stoga znatno brže od nasumičnog čitanja i pisanja.

Kafka se može konfigurisati za asinhrono ispiranje na disk (async flush) bez uključivanja sinhronog ispiranja. Asinhrono ispiranje ne čeka da se poruke upišu na disk pre nego što pošalje ACK o isporuci poruke, pa se na taj način povećava propusnost slanja poruka i smanjuje latencija zahteva.

"Zero Copy"

Tokom tradicionalnog IO toka, podatke je prvo potrebno kopirati u kernel bafer, zatim iz kernel bafera u korisnički prostor, a nakon što aplikacija završi obradu, ponovo ih kopirati nazad u kernel bafer.

Tok ovog procesa izaziva višestruko kopiranje podataka.

Kako bi se smanjilo nepotrebno kopiranje, Kafka se oslanja na Sendfile sistemski poziv koji pruža Linux kernel.

U Sendfile metodi, ulaz i izlaz podataka se obavljaju u kernel baferu, bez potrebe za kopiranjem u korisnički prostor radi obrade, čime se izbegava višestruko kopiranje podataka.

U konkretnoj primeni, Kafka sve poruke čuva u zasebnom fajlu, a pri isporuci poruka direktno šalje fajl preko metode Sendfile, što smanjuje preklapanja konteksta i drastično povećava performanse.

"MMAP tehnologija"

Pored Sendfile-a, postoji još jedna tehnologija kojom se ostvaruje zero copy, a to su Memory Mapped Files.

Kafka koristi Memory Mapped Files za mapiranje memorije. Operacije nad fajlovima preko Memory Mapped Files ne koriste write/read, već direktno rade nad memorijskim adresama. Ako se pozove operacija read nad fajlom, podaci se prvo pročitaju u kernel prostor, a zatim kopiraju u korisnički prostor; međutim, MMAP omogućava da se fajl direktno mapira u memoriju korisničkog prostora, čime se eliminiše trošak kopiranja iz korisničkog u kernel prostor.

Podaci koje proizvede Producer, kada se trajno upisuju na Broker, koriste mmap mapiranje fajlova radi brzog sekvencijalnog upisivanja.

Consumer, kada čita podatke iz Broker-a, koristi sendfile: disk fajl se pročita u OS kernel bafer, a zatim direktno prosledi u socket bafer radi slanja preko mreže.

"Grupno slanje i čitanje"

Grupisane operacije Kafke obuhvataju grupno upisivanje, grupno objavljivanje, itd. Pri isporuci poruka Kafka ih prvo kešira, a zatim šalje u grupama.

Isto tako, potrošač pri konzumiranju poruka ne obrađuje ih jednu po jednu, već ih povlači u grupama, što povećava brzinu obrade poruka.

"Kompresija podataka"

Kafka takođe podržava kompresiju skupova poruka. Producer može izvršiti kompresiju skupa poruka u GZIP ili Snappy formatu.

Koris od kompresije jeste smanjenje količine podataka koji se prenose i rasteraćenje mrežnog prenosa.

Nakon kompresije od strane Producera, na strani Consumer-a potrebna je dekompresija. Iako se povećava rad CPU-a, u obradi velikih količina podataka usko grlo je mreža, a ne CPU, pa je ovaj trošak opravdan.

"Mehanizam particija"

Sadržaj teme u Kafci može se podeliti u više particija, a svaka particija se deli na više segmenata (segment), pa je svaka operacija zapravo nad malim delom — vrlo lako — i povećava mogućnost paralelnih operacija.

Najčešća pitanja na intervjuu

"Da li je Kafka Push ili Pull model?"

Prvo pitanje koje je Kafka postavila bilo je: da li Consumer treba da povlači poruke iz Broker-a ili Broker da gura poruke ka Consumer-u.

Po ovom pitanju, Kafka prati tradicionalan dizajn koji dele skoro svi sistemi poruka: Producer gura poruke ka Broker-u, a Consumer povlači poruke iz Broker-a.

U Push modelu, Broker određuje brzinu slanja poruka, što stvara poteškoće pri obradi potrošača različitih brzina konzumiranja.

Sistemi poruka teže ka tome da Consumer konzumira poruke što brže i uz maksimalnu brzinu. U Push modelu, kada brzina slanja znatno premašuje brzinu konzumiranja, Consumer može da se sruši.

❝

Kafka Producer i Consumer koriste Push-and-Pull model: Producer gura (Push) poruke ka Broker-u, a Consumer povlači (Pull) poruke iz Broker-a.

❞

Jedna prednost Pull modela jeste u tome što Consumer samostalno odlučuje da li će u grupama povlačiti podatke iz Broker-a.

Mana Pull modela jeste u tome što, ako Broker nema poruke za konzumiranje, Consumer zapada u neprekidno ispitivanje (polling) u petlji, sve dok ne stigne nova poruka.

"Kako Kafka obezbeđuje visoku dostupnost?"

Intervju: Kako Kafka obezbeđuje visoku dostupnost? Sa slikama i dokazima

"Scenariji korišćenja Kafke"

Scenariji u kojima se Kafka koristi u industriji:

❝

Asinhrona komunikacija

❞

Middleware za poruke najviše se koristi u asinhronoj komunikaciji. U mnogim poslovnim tokovima, ako se svi koraci obavljaju sinhrono, to može dovesti do toga da osnovni tok potraje vrlo dugo. Još važnije, ako svi koraci idu sinhrono, neuspeh nekog neključnog koraka može srušiti čitav osnovni tok. Zato Kafka u mnogim poslovnim tokovima ima ulogu asinhrone komunikacije.

❝

Sinhronizacija logova

❞

Masivni distribuirani sistemi obuhvataju mnogo mašina razasutih po različitim server-sobama, a jedan od očitih problema koje donosi distribuirani sistem jeste to što pregled, praćenje i analiza poslovnih logova postaju vrlo teški. Za sisteme sa klasterom od stotina mašina, pretraživanje logova u produkciji je noćna mora.

Da bi se odgovorilo na ovaj scenario, javljaju se objedinjeni sistemi za logove. Podaci logova su masivni, pa da ne bi opteretili sistem, najčešće se koristi asinhrono slanje. Zbog svog visokog protoka, Kafka je ovde našla odličnu primenu u obradi logova.

❝

Realno vreme izračunavanje

❞

Sa porastom količine podataka, oflajn izračunavanje postaje sve sporije i teško može da zadovolji zahteve za realno vreme u određenim scenarijima, pa mnoga rešenja uvode realno-vremensko izračunavanje.

Često, čak i kada su podaci masivni, želimo da odmah pogledamo neke pokazatelje, pa se javlja realno-vremensko strujno izračunavanje.

Realno-vremensko strujno izračunavanje ima dve osobine: jedna je realno vreme — u svakom trenutku možemo videti podatke; druga je strujno — podaci teku.

Izvor: https://mp.weixin.qq.com/s?__biz=MzUyOTg1OTkyMA==&mid=2247487063&idx=1&sn=d7c65359630a67695a079d62fce989c0&chksm=fa5bda68cd2c537ed32003d5a98705e85c6b7a47afe0f837cebd27a5f33a8a11231d3eb9c1c5&scene=27#wechat_redirect, poreklo: Yuebaan Feiyu, obradio: Chenmo Wang Er