Apache Kafka: Message Flow, Failure və Consumer Processing
Əvvəlki yazıda Apache Kafka-nın nə olduğu, topic, partition, offset, broker, replication, producer, consumer və consumer group kimi əsas anlayışlardan danışmışdıq. LINK
Bu ilkin anlayışları bildikdən sonra bilmək lazımdır ki production-da Kafka istifadə etdikcə məsələlər biraz mürəkkəb olmağa başlayır. Producer Kafka-ya bir mesaj göndərdi. Kafka həmin mesajı hansı partition-a yazmalıdır? Producer mesajın uğurla yazıldığını haradan bilir? Broker mesajı qəbul etdikdən dərhal sonra çöksə nə baş verəcək? eyni mesajın iki dəfə oxunması nəyə səbəb ola bilər? və s.
Bu yazıda bir mesajın producer-dən çıxıb consumer tərəfindən oxunmasına qədər hansı proseslərdən keçdiyinə baxaq. Bunun üçün bank sistemlərində baş verə biləcək bir nümunəyə baxaq. Tutaq ki, Transfer Service var və istifadəçi A hesabından B hesabına 100 AZN pul köçürür. Transfer uğurla tamamlandıqdan sonra digər sistemlərin bundan xəbərdar olması üçün transfer-completed topic-nə aşağıdakı mesaj göndərilir.
{
"eventId": "8d53d983-79ef-4e90-8b84-4358976a91f3",
"transferId": "TRX-981723",
"sourceAccountId": "ACC-1001",
"destinationAccountId": "ACC-2001",
"amount": 100,
"currency": "AZN"
}Notification Service bu mesajı oxuyub istifadəçiyə notification göndərə, Reporting Service hesabat yarada, Fraud Service prosesi analiz edə bilər. Hələki hər şey sadə görünür. Ancaq producer.send(…) metodu çağırıldıqdan sonra mesaj birbaşa broker-ə gedib diskə yazılmır. Bu prosesin arxasında partition seçilməsi, serialization, batching, replication, acknowledgement, retry və s. başqa proseslər baş verir.
Əvvəlki yazı 2022-ci ildə yazıldığı üçün Kafka cluster-in idarə olunmasında Zookeeper-dən danışmışdıq. Kafka-nın yeni versiyalarında artıq Zookeeper istifadə olunmur. Kafka 4.x yalnız KRaft(Kafka Raft) mode ilə işləyir. Hansı broker-lərin cluster-də olması, hansı partition-un hansı broker-də yerləşməsi, leader-in hansı broker olması kimi metadata məlumatlarını Kafka-nın öz controller-ləri tərəfindən həyata keçirilir artıq. Production kimi kritik environment-lərdə broker və controllerlərin ayrı server-lərdə saxlanılması məsləhət görülür. Adətən controller-lər də say olaraq 3 və ya 5 olur. 3 controller olduqda bir controller-in sıradan çıxması, 5 olduqda iki controller-in sıradan çıxması böyük problem yaratmır.
Java-da Kafka-ya mesaj göndərərkən adətən KafkaProducer və ya Spring istifadə etdiyimiz halda KafkaTemplate işlədirik.Aşağıdakı kimi bir message göndərdiyimizi düşünək:
ProducerRecord<String, TransferCompletedEvent> record =
new ProducerRecord<>(
"transfer-completed",
transfer.getSourceAccountId(),
event
);
producer.send(record);
// topic = transfer-completed
// key = ACC-1001
// value = TransferCompletedEventsend() methodu çağırıldıqdan sonra producer ilk olaraq key və value dəyərlərini byte şəklinə çevirməlidir, cünki Kafka broker bizim Java object-in nə olduğunu bilmir. Producer-də key.serializer və value.serializer bunun üçün istifadə olunur. Yəni bizim göndərdiyimiz TransferCompletedEvent obyekti şəbəkə üzərindən byte array kimi gedir. Producer-in send() metodu async işləyir. Message əvvəl producer-in daxili buffer-nə düşür, daha sonra backgroundda olan I/O thread həmin message-ləri broker-lərə göndərir. Yəni send() methodunun çağırılması message-in Kafka-ya uğurla yazıldığı demək deyil.
producer.send(record);
log.info("Transfer event successfully sent");bu log düzgün deyil. Çünki biz sadəcə send() metodunu çağırmışıq, broker-dən acknowledgement hələ gəlməmiş ola bilər. Ona görə də success və ya fail olduğunu callback vasitəsilə yoxlaya bilərik.
producer.send(record, (metadata, exception) -> {
if (exception != null) {
log.error("Kafka message could not be sent", exception);
return;
}
log.info(
"Message sent. partition={}, offset={}",
metadata.partition(), metadata.offset());
});Callback daxilində kompleks database query (hansı ki icrası uzun vaxt alır), başqa external servislərə HTTP request, icrası uzun vaxt alan logic-lər yazmaq düzgün deyil. Bu thread-i blokladıqda digər message-lərin göndərilməsi də gecikəcək. Bank kimi saniyədə minlərlə və ya on minlərlə event göndərilən sistemdə belə kiçik görünən bir səhv producer throughput-un ciddi şəkildə aşağı düşməsinə səbəb ola bilər.
Serialization-dan sonra əsas məsələlərdən biri mesajın hansı partition-a göndəriləcəyidir. Əvvəlki yazıda qeyd etdiyimiz kimi partition Kafka-nın paralel işləməsinin əsas səbəblərindən biridir. Bir topic bir neçə partition-dan ibarət ola bilər. Producer mesaj göndərən zaman partition-u özümüz təyin edə bilərik, amma adətən partition-u manual təyin etmək lazım olmur. Əgər key varsa Kafka həmin key-in hash dəyərinə əsasən partition seçir. Eyni key ilə gələn mesajlar eyni partition-a düşür. Eyni key verəərək mesajları eyni partition-a göndərməyin əsas səbəblərindən biri ordering-dir. Kafka topic üzrə message ordering zəmanəti vermir. Ordering yalnız bir partition daxilində qorunur. Məsələn bizə aid account üzrə debit/credit event-lərinin hamısının eyni partition-a düşməsi lazımdır. Buna görə biz key olaraq accountId istifadə edə bilərik. Consumer həmin partition-u sıralı şəkildə oxuduğu üçün ordering qorunacaq. Digər tərəfdən bütün event-ləri bir key altında toplamaq da düzgün deyil. Betə etsək əgər 30 partition olsa belə faktiki olaraq bir partition işləyəcək. Digər partition-lar boş qalacaq və bütün yük bir partition üzərinə düşəcək. Bu problemə hot partition deyilir.
Başqa bir vacib məqam partition sayının sonradan artırılmasıdır. Key əsasında partition seçimi mövcud partition sayından asılı olur və biz partition sayını artırıqda topic-in partition sayı dəyişir və yeni mesajlar başqa partition-a düşə bilər. Buna görə ordering-in kritik olduğu sistemlərdə partition sayını dəyişdirmək əvvəlcədən düşünülməli olan architecture qərarıdır.
Producer-in əsas performans qazandıran səbəblərindən biri message-ləri bir-bir göndərmək əvəzinə batch şəklində göndərə bilməsidir. Message-lərin hamısını ayrı-ayrı network request kimi getməsi əvəzinə Kafka onları eyni partition üçün bir batch daxilində toplayıb göndərir. Burada əasas iki parametr bizə lazımdır : batch.size və linger.ms
batch.size bir partition üçün yaradılan batch-in ölçüsünü müəyyən edir. Hazırkı Kafka producer-də default olaraq 16 KB-dır.
linger.ms isə producer-in batch-ə başqa message-lərin də gəlməsini gözləyə biləcəyi müddətdir. Kafka 4.x-də default 5 ms olur. batch.size dolarsa linger.ms bitməsini gözləmədən batch göndərilə bilər.
Producer message-ləri batch şəklində saxladığı üçün memory də istifadə edir. Bunun üçün buffer.memory parametri var. Məsələn Kafka saniyədə 2000 message qəbul edə bilir amma Applicaiton 10_000 message göndərir. Bu aradakı fərq producer buffer-də yığılır. Buffer dolduqdan sonra producer-in asynchronous olduğunu fikirəşib application-ın problemsiz işləməsini gözləmək doğru deyil. send() methodu buffer dolduğu üçün block ola bilər və max.block.ms limitinə çatdıqda xəta baş verə bilər. Bu Kafka istifadə etdiyimiz servisin özünə də təsir edə bilər. Məsələn Payment Service Kafka-ya event göndərir və Kafka cluster-də problem yaranıb. Və əgər producer backpressure, timeout və error handling düzgün dizayn olunmayıbsa Kafka problemi biraz sonra Payment Service-in thread-lərini bloklayacaq. Distributed sistemlərdə bir sistemin problemi rahat şəkildə digər sistemə ötürülə bilir.
Producer batch-ləri broker-ə göndərməzdən əvvəl compress də edə bilər. Kafka gzip, snappy, lz4, zstd kimi compression tiplərini dəstəkləyir, ətraflı : https://www.confluent.io/blog/apache-kafka-message-compression/
Daha çox compression etmək network yükünü azaldır amma CPU yükünü artırır. Message-lər ölçüse böyükdürsə və network throughput problemdirsə compression faydalıdır. Amma CPU usage onsuz da çoxdursa compression problem yarada bilər. Yəni Kafkada performance artlrmaq istəsək Network, CPU, memory, disk, throughput, latency və s. hamısını birlikdə nəzərə almalıyıq.
Producer configuration-da həmişə belə bir parametr görürük:
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092Buradakı broker-lər producer-in bütün message-ləri göndərdiyi broker-lər deyil. Bunlar producer-in Kafka cluster-i ilk dəfə tapması üçündür. Producer bootstrap broker-lərdən birinə qoşulur və cluster metadata-sını (hansı topic-lərin olduğunu, neçə partition olduğunu, leader hansı olduğu və s.) əldə edir. Bundan sonra producer hansı partition-a message göndərirsə birbaşa həmin partition-un leader broker-inə request göndərə bilir. Best practice olaraq bootstrap server list-də birdən çox broker qeyd etmək lazımdır. Bir bootstrap broker işləməsə producer digəri vasitəsilə cluster metadata-sını ala bilir.
Producer partition-u müəyyən edir və message-in leader broker-inə request göndərir. Message Broker-ə gəldikdən sonra partition-un log-na append edilir. Kafka log structure istifadə edir. Yeni message əvvəlki message-i update etmir, log-un sonuna əlavə olunur. Message yazıldıqdan sonra partition daxilində offset alır. Əvvəlki yazıda da qeyd etdiyimiz kimi offset global deyil. Partition 0 / Offset 100 və Partition 1/ Offset 100 fərqli message-lərdir. Offset-in mənası yalnız aid olduğu partition daxilində var. İndi bu vəziyyətdə əgər leader broker çöksə Kafka-nın ən vacib anlayışlarından biri olan replication işə düşür. Tutaq ki Topic partition=6 və replication.factor = 3 olaraq yaradılıb. Bu o deməkdir ki, Partition 2 üç fərqli broker-də (1 Leader və 2 In-Sync Replicas) saxlanıla bilər. Producer write request-i leader olan Broker-ə göndərir. In-Sync Replica-lar leader-dən məlumatları götürərərk sinxronlaşdırırlar. ISR(In-Sync Replicas)- Leader ilə sinxron vəziyyətdə olan replica-lar adlanır.
Broker-lərdən biri gecikməyə başlayarsa və leader ilə sinxron qala bilməzsə ISR-dən çıxarıla bilər. Bu o demək deyil ki həmin Broker artıq replica deyil. Sadəcə hazırda in-sync replica hesab edilmir. Broker çökdükdə cluster başqa uyğun replica-nı leader seçə bilər. Kafka 4.x-də bu metadata və leader election prosesinin idarə olunması KRaft controller-ləri tərəfindən yerinə yetirilir. Burada production üçün başqa vacib bir məsələ varki 3 replica yaratmaq kifayət deyil, həmin replica-ların fiziki olaraq harada yerləşdiyinə də baxmaq lazımdır. Məsələn hamısı eyni zone-dadırsa həmin zone tam sıradan çıxdıqda replication factor 3 olmasının bizə faydası yoxdur. Kafka rack awareness vasitəsilə eyni partition-un replica-larını fərqli rack-lər arasında bölə bilir.
Producer mesajın uğurla yazıldığını bildirən acks parametri var. Əvvəlki məqalədə bunu izah etmişdim. acks=all producer ən güclü durability səviyyəsidir. Leader message-i qəbul edir və mövcud ISR-lərin write-i acknowledgement etməsini gözləyir. Hazırkı Kafka producer-də acks default olaraq all-dır. Burada vacib bir məsələ var, təkcə acks=all etmək tam kifayət etmir. Burada min.insync.replicas də nəzərə alınmalıdır. 2 Broker olub min.insync.replicas = 3 olsa Producer NotEnoughReplicas kimi bir exception verə bilər. Bank kimi kritik sistemlərdə istifadə olunan configuration belə ola bilər:
replication.factor = 3
min.insync.replicas = 2
acks = allBu configuration iki broker işlədiyi müddətdə write etməyə imkan verir, amma bir replica qaldıqda write operation-ları dayandırır.
Başqa bir vəziyyətə baxaq. Producer message göndərir, Leader message-i yazır və ACK göndərir. Ancaq network problemi səbəbiylə ACK producer-ə çatmır. Producer heç nə bilmədiyi üçün iki ehtimal var: Ya message heç yazılmayıb ya da message yazılıb amma ack gəlməyib. Producer message-i yenidən göndərsə və birinci message həqiqətən yazılmışdısa iki eyni message yarana bilər. Göndərməsə və birinci message həqiqətən yazılmamışdısa message itə bilər. Kafka producer buna görə retriable error-larda avtomatik retry edə bilir. Hazırkı Kafka producer-də retries default dəyəri yüksəkdir və Kafka documentation retry sayını manual idarə etməkdənsə delivery.timeout.ms vasitəsilə idarə etməyi tövsiyə edir. Default 2 dəqiqədir. Burada application-da özümüz əlavə retry yazarkən diqqətli olmalıyıq. Məsələn:
try {
kafkaTemplate.send(...).get();
} catch (Exception ex) {
kafkaTemplate.send(...);
}Birinci send Kafka-da uğurlu olub, sadəcə cavabı bizə çatmayıbsa application səviyyəsində ikinci dəfə göndərdiyimiz message yeni send kimi qəbul oluna bilər. Producer-in daxili retry mexanizmi ilə application səviyyəsində bizim exception handle edərək yenidən send() methodu çağırmağımız eyni şey deyil.
Tutaq ki eyni partition-a iki batch göndərilib: Batch A və Batch B . Batch A fail olur. Batch B isə uğurla yazılır. Sonra Batch A retry olunur. Nəticədə Batch B sonra Batch A yazılır. Yəni retry message ordering-ə təsir edə bilər. Retry problemini həll etmək üçün Kafka-da ən vacib mexanizmlərdən biri Idempotent Producer-dir. Idempotence sadə dildə eyni əməliyyat bir neçə dəfə təkrar olunsa belə nəticənin bir dəfə baş verməsini təmin etməkdir. Bu sayədə Producer eyni message-i retry etdikdə broker yeni message yox, əvvəlki message-in olduğunu müəyyən edə bilir. Bunun üçün Kafka producer id və message batch-ləri üçün sequence məlumatlarından istifadə edir. Kafka-nın message formatında producerId, producerEpoch və baseSequence kimi məlumatların saxlanmasının səbəblərindən biri də budur. Hazırkı Kafka producer-də idempotence default olaraq aktivdir, əgər onunla konflikt yaradan configuration yazılmayıbsa.
Idempotence üçün əsas şərtlər:
acks=all
retries > 0
max.in.flight.requests.per.connection <= 5
Bir şeyi unutmamaq lazımdır ki, Idempotent Producer bizim bütün sistemimizi idempotent etmir. Kafka producer-in idempotence mexanizmi producer retry nəticəsində Kafka log-da yaranacaq duplicate-lərin qarşısını alır. Amma əgər application Kafka-ya message göndərdi və sonra application özü yenidən yeni send() etdisə burada artıq idempotence mexanizmi işləmir. Idempotence producer session daxilində öz retry-ləri üçün qarantiya verir. Bundan əlavə producer idempotence consumer-in eyni business event-i iki dəfə process etməyəcəyinə də qarantiya vermir. Bu hissəyə consumer-dən danışdıqda yenidən qayıdacağıq. Buraya qədər message-in Producer-dan çıxıb Kafka broker-də saxlanmasına qədər olan hissəyə baxdıq. Artıq mesajımız Kafka-dadır.
İndi isə Consumer-in bu message-i necə oxuduğuna və oxuduqdan sonra hansı problemlərin yarana biləcəyinə baxaq.
Consumer Kafka-dan message oxuyan tərəfdir. Producer-də olduğu kimi consumer də əvvəl bootstrap.servers vasitəsilə Kafka cluster-ə qoşulur və lazım olan metadata məlumatlarını əldə edir. Consumer topic-ə subscribe olduqda group.id vasitəsilə hansı consumer group-a aid olduğunu bildirir. Tutaq ki transfer-completed topic-də 6 partition var və Fraud Service-dən 3 instance işləyir. Bu 3 instance eyni group.id istifadə edirsə eyni consumer group-a aid olurlar və partition-lar onların arasında bölünür.
Consumer 1 -> Partition 0, 1
Consumer 2 -> Partition 2, 3
Consumer 3 -> Partition 4, 5Bir consumer group daxilində bir partition eyni anda yalnız bir consumer tərəfindən consume olunur. Ona görə də 6 partition olan topic üçün eyni consumer group-da 10 consumer qaldırsaq maksimum 6 consumer işləyəcək, qalan 4 consumer-in oxuyacağı partition olmayacaq. Application-da sadəcə instance sayını artırmaq hər zaman throughput-u artırmır.
Get Mahammad Eminov’s stories in your inbox
Join Medium for free to get updates from this writer.
Consumer broker-in message-i ona göndərməsini gözləmir. Consumer özü broker-ə fetch request göndərərək message-ləri götürür, yəni pull model ilə işləyir. Java KafkaConsumer istifadə etdikdə bunun əsas hissəsi poll() methodudur.
ConsumerRecords<String, TransferCompletedEvent> records =
consumer.poll(Duration.ofMillis(100));poll() bir və ya bir neçə record qaytara bilər. Consumer hansı partition-ların ona assign olunduğunu bilir və həmin partition-ların leader broker-lərindən message-ləri fetch edir. Burada max.poll.records parametri var. Hazırkı Kafka-da default olaraq bir poll() maksimum 500 record qaytarır. Bu parametr broker-dən bir dəfəyə nə qədər data fetch olunmasını təyin etmir, consumer-in artıq fetch edib saxladığı record-lardan bir poll() zamanı application-a maksimum neçə record veriləcəyini təyin edir. poll() bizə 500 transfer event qaytarıbsa application həmin 500 record-u process etməlidir. Burada artıq Kafka consumer-də ən vacib məsələlərdən biri olan offset mövzusuna gəlirik. Əvvəlki məqalədə qeyd etdiyimiz kimi hər message-in partition daxilində offset-i var. Burada iki məqam var:
current position və committed offset
Current position consumer-in hazırda partition-un hansı hissəsinə qədər gəldiyini göstərir.
Committed offset isə Kafka-da saxlanılan və consumer restart olduqda haradan davam etməli olduğunu göstərən dəyərdir.
Məsələn Consumer Offset 100, 101 və 102 message-lərini oxuyub. Consumer-in növbəti oxuyacağı message Offset 103-dür. Commit olunan offset də adətən 103 olur. Yəni commit etdiyimiz offset son process etdiyimiz message-in offset-i yox, növbəti oxumalı olduğumuz offset-dir.
Consumer offset məlumatlarını Kafka-nın internal __consumer_offsets topic-ində saxlayır. Consumer restart olduqda həmin group üçün son committed offset tapılır və consumer oradan davam edir. Məsələn, Committed Offset = 103 olduqda Consumer restart olduqda bilir ki 100, 101, 102 artıq əvvəlki consumer tərəfindən process olunub və növbəti 103-dən davam etməlidir.
Consumer message-i process etməmiş commit etsək bir problem yaranır, process etdikdən sonra commit etsək başqa problem yarana bilər. Tutaq ki consumer Offset 100 olan TransferCompleted event-i oxudu. Əgər əvvəlcə offset-i commit etsək və sonra Business logic-də hər hansı problem baş verdi və ya application çökdüsə, Kafka-da bu message artıq proses olunmuş kimi saxlanılır. Consumer yenidən qalxdıqda növbəti offset-dən davam edəcək. Amma bizim business logic tamamlanmayıb. Bu At Most Once-dır. Message maksimum bir dəfə process olunur amma müəyyən failure vəziyyətində ümumiyyətlə process olunmaya bilər.
Başqa variantda əvvəl process edirik, sonra offset-i commit edirik. Bu daha düzgün görünür. Amma belə bir vəziyyət ola bilərki Message oxunur, sonra business logic tamamlanır və offset commit hissəsinə çatmamış application çökür. Consumer yenidən qalxanda son committed offset-ə baxacaq və həmin message-i yenidən oxuyacaq. Bu At Least Once-dır. Message itmir amma eyni message bir neçə dəfə process oluna bilər. Bank kimi kritik sistemlərdə çox vaxt message itirməkdənsə duplicate message qəbul etmək daha düzgün seçimdir. Amma o haldaki application duplicate message-i düzgün handle etməlidir. Məsələn bizim TransferCompleted event:
{
"eventId": "8d53d983-79ef-4e90-8b84-4358976a91f3",
"transferId": "TRX-981723",
"amount": 100
}consumer tərəfindən iki dəfə gəldi. Əgər Notification Service-dirsə istifadəçiyə eyni notification iki dəfə getməsi pisdir amma o qədər də kritik deyil. Amma həmin consumer message gəldikdə hesabdan 100 AZN çıxırsa problem kritikdir. Ona görə də At Least Once istifadə etdiyimiz sistemdə Consumer-in idempotent olması çox vacibdir. Bunun yollarından biri eventId kimi dəyər saxlamaqdır. eventId görə yoxlama aparmaq olarki artıq process olunubsa business logic-i ikinci dəfə yerinə yetirməsin.
Producer idempotence producer-in öz retry-ləri nəticəsində Kafka log-da duplicate yaranmasının qarşısını alır.
Consumer idempotency isə eyni business event Kafka-dan bir neçə dəfə gəldikdə application-ın eyni əməliyyatı bir neçə dəfə etməsinin qarşısını alır.
eventId yoxlaması aparan zaman application çökə bilər və ya concurrency səbəbiylə eyni event üçün race condition yarana bilər. Ona görə daha düzgün yollardan biri eventId üçün database-də unique constraint olması və prosesin eyni database transaction daxilində olmasıdır (burada database lock da istifadə oluna bilər).
Əgər consumer başqa external sistemə HTTP request göndərirsə məsələ biraz daha çətin olur. Məsələn Kafka-dan PaymentRequested event oxuyub başqa payment provider-ə request göndəririk. Consumer request-i göndərdi, provider payment-i etdi amma bizim service response gəlməmiş çökdü. Message yenidən consume olduqda həmin request yenidən göndərilə bilər. Bu halda external sistem idempotency key dəstəkləyirsə eventId və ya paymentId kimi bir id-ni request ilə göndərmək lazımdır. Yəni Kafka-da duplicate problemi yalnız Kafka config ilə həll olunan məsələ deyil.
Plain KafkaConsumer-da enable.auto.commit default olaraq true-dur. Bu halda consumer offset-ləri müəyyən interval ilə avtomatik commit edir. auto.commit.interval.ms default olaraq 5 saniyədir. Ancaq auto commit-in olması message-in business logic tərəfindən uğurla process olunduğu mənasına gəlmir. Yəni kritik sistemlərdə auto commit istifadə etmək risklidir.
Manual commit etdikdə commitSync() və commitAsync() kimi methodlardan istifadə edə bilərik. commitSync() broker-dən commit cavabını gözləyir. Hər record-dan sonra commitSync() çağırmaq throughput-a mənfi təsir edə bilər. Hər message üçün ayrıca commit request etmək də lazımsız əlavə network operation yarada bilər. Bir batch process edib sonra commit etmək performans baxımından daha yaxşı olar, amma batch böyüdükcə failure zamanı yenidən process oluna biləcək message sayı da artır. Məsələn poll() 500 record qaytardı, 500 record process olundu və offset commit etməmiş application çökdü. Consumer restart olduqda bu 500 record-un müəyyən hissəsi və ya hamısı yenidən process oluna bilər. Burada yenə throughput və duplicate processing ehtimalı arasında qərar verməli oluruq.
Spring Kafka istifadə etdikdə isə bir məqamı nəzərə almaq lazımdır. Kafka client-in enable.auto.commit default-u true olsa da Spring Kafka listener bunu explicit olaraq verməsək false edir və offset commit-i öz AckMode mexanizmi vasitəsilə idarə edir. Default AckMode BATCH-dır. Yəni @KafkaListener istifadə etdikdə documentation-dakı auto commit məsələsi Spring listener-ə aid olmur.
İndi Exactly Once məsələsinə baxaq. Kafka daxilində bir topic-dən message oxuyub başqa Kafka topic-nə message yazdığımız flow-da Kafka transactions vasitəsilə consumer offset və producer write əməliyyatı birlikdə transaction daxilində idarə oluna bilər. Producer transaction açır, message Topic B-yə yazır və Consumer-in offset commit də həmin transaction-a daxil edir. Hər hansı problem olarsa olarsa ikisi də commit olunmur. Bu flow-da consumer tərəfdə isolation.level = read_committed istifadə etməklə fail olmuş transaction daxilində yazılan message-lərin consumer-ə çatmasının qarşısını almaq olar. ( Kafka-da isolation.level default olaraq read_uncommitted-dir. ) Amma Consumer Kafka-dan message oxuyub Oracle, PostgreSQL və ya başqa external sistemdə əməliyyat edirsə exactly once burada işləmir. PostgreSQL update uğurla oldu, offset commit olunmadan application çökdüsə message yenidən gələcək.
Consumer Group daxilində başqa bir vacib məsələ Rebalance-dır. Tutaq ki 6 partition və 3 consumer var:
Consumer A -> P1, P2
Consumer B -> P3, P4
Consumer C -> P5, P6
Əgər Consumer B çökərsə P2 və P3 partition-ları qalan consumer-lər arasında yenidən assign edilməlidir. Bu proses rebalance adlanır. Yeni consumer group-a qoşulduqda, consumer group-dan çıxdıqda, consumer crash olduqda assignment prosesi gedir. Rebalance əslində normal prosesdir amma çox tez-tez baş verməsi problemdir. Rebalance zamanı partition ownership dəyişdiyi üçün processing yavaşlaya bilər və offset-lər düzgün idarə olunmazsa duplicate processing də yarana bilər.
Tutaq ki Consumer B P2-də Offset 500 olan message-i process edir. Business operation tamamlandı amma offset commit olunmamış rebalance başladı və P2 Consumer A-ya keçdi. Consumer A son committed offset-dən başlayacaq və həmin message yenidən process oluna bilər. Yenə eyni yerə gəlirik: At Least Once olan sistemdə Consumer idempotent olmalıdır.
Consumer group-un consumer-in işləyib işləmədiyini bilməsi üçün heartbeat mexanizmi var. Consumer müəyyən aralıqlarla Group Coordinator-a heartbeat göndərir. session.timeout.ms müddətində heartbeat gəlməzsə consumer group-dan çıxarılır və rebalance prosesi başlayır. Kafka 4.x-də yeni Consumer Rebalance Protocol da var.
Ətraflı: https://kafka.apache.org/43/operations/consumer-rebalance-protocol/
Amma consumer-in işləyib-işləmədiyini bildirən təkcə heartbeat göndərməsi deyil. Consumer message-ləri vaxtında poll() etməlidir. Burada max.poll.interval.ms vacib parametrdir. Default olaraq 5 dəqiqədir. Məsələn max.poll.records=500-dür. poll() 500 message qaytardı və hər message-in processing-i tutaq ki 1 saniyə çəkir. 500 saniyə təxminən 8 dəqiqədən çoxdur. max.poll.interval.ms isə 5 dəqiqədir. Consumer hələ əvvəlki batch-i process etdiyi halda Kafka onu group-dan çıxara bilər və rebalance baş verər. Bir müddətdən sonra consumer offset commit etmək istədikdə artıq həmin partition onun olmayacaq və commit fail olacaq. Bir neçə həll yolu ola bilər. max.poll.records azaltmaq olar, processing logic-i sürətləndirmək olar, database query-ləri optimizasiya etmək olar və ya max.poll.interval.ms business processing vaxtına uyğun yazıla bilər.
Digər bir məsələ message-ləri poll edib sürətli processing üçün thread pool-a atmaq hərzaman düzgün deyil. Məsələn eyni account-a aid event-ləri eyni partition-a göndərmişdik, cünki ordering lazım idi. Əgər consumer bunları götürüb müxtəlif thread-lərlə parallel process etsə bizim Kafka partition ilə qoruduğumuz ordering application daxilində pozular. Yəni producer tərəfində key seçməklə ordering qururuq, digər consumer tərəfdə isə multithreading ilə message-ləri proses edib həmin ordering-i poza bilərik.
Consumer restart-ları da rebalance yarada bilər. Production-da deployment zamanı məsələn 20 consumer instance eyni anda restart olursa group bir neçə dəfə rebalance ola bilər. group.instance.id istifadə etdikdə consumer static member kimi tanına bilir və qısa müddətli restart-ların yaratdığı lazımsız rebalance-ları müəyyən hallarda azaltmaq mümkündür.
Consumer Lag məsələsinə baxaq. Producer topic-ə message yazmağa davam edir, Consumer isə həmin message-ləri oxuyur. Consumer təxminən 500 message geridə olduqda bu lag adlanır. Lag-ın olması həmişə problem deyil. Yəni Producer müəyyən vaxt aralığında çox message yaza bilər və consumer biraz sonra onları oxuya bilər. Əsas problem lag davamlı olaraq böyüyürsə başlayır. Yəni Consumer producer-in sürətinə çata bilmir. Bunun çox səbəbi ola bilər. Consumer-in business logic-i yavaşdır, database gec cavab verir, external service latency yaradır, rebalance baş verir, consumer sayı azdır və s..
Bank sistemi kimi kritik sistemdə sadəcə “consumer up”-dır monitor etmək kifayət deyil. Məsələn Fraud Service kart əməliyyatlarını 20 dəqiqə gecikmə ilə yoxlayırsa health endpoint-in 200 qaytarmasının əhəmiyyəti yoxdur. Ona görə lag üçün alert-lər business requirement-ə görə qurulmalıdır. Bəzi topic üçün 10_000 lag normal ola bilər, başqa topic üçün 100 belə kritik ola bilər. Lag artdıqda ilk ağla gələn “consumer sayını artıraq” olur. Ama əvvəl dediyimiz kimi bunun da limiti partition sayından asılıdır.
Başqa production problemi müəyyən bir message-in hər dəfə fail olmasıdır. Məsələn consumer message-i oxuyur və Offset 502 olan message fail verir və bizim retry mexanizmimiz Offset 502-ni sonsuz retry edir. Bu halda həmin partition-da 503, 504, 505 və sonrakı message-lərə növbə gəlməyəcək. Producer isə yeni message göndərməyə davam etdikcə lag böyüyəcək. Məsələn database 2 saniyə unavailable oldu, external service timeout verdi və s. belə halda retry məntiqlidir. Amma message-in özü səhvdirsə istənilən qədər retry etsək də bir mənası yoxdur. Müəyyən sayda retry-dan sonra failure davam edirsə message-i ayrıca Dead Letter Topic-ə göndərmək lazımdır. Və sonra səbəbi həll olunduqdan sonra replay mexanizmi olmalıdır.
Deserialization error da bəzən baş verən problemlərdən olur. Producer modeli dəyişdi amma Consumer yeni modeli deserialize edə bilmədiyi üçün fail olacaq. Ona görə event model dəyişikliklərində backward compatibility, producer və consumer deployment sırasını nəzərə almaq vacibdir.
Başqa vacib parametr auto.offset.reset-dir. Bu parametr çox vaxt səhv başa düşülür. auto.offset.reset=earliest yazmağımız consumer hər restart olduqda topic-i əvvəldən oxuyacaq demək deyil. auto.offset.reset yalnız consumer group üçün committed offset olmadıqda istifadə olunur. Məsələn yeni group.id yaradılıb və bu group heç vaxt topic oxumayıb. earliest olsa mövcud ən köhnə offset-dən başlayacaq oxumağa.
latest olsa hazırda log-un sonundan başlayacaq və əvvəlki message-ləri oxumayacaq və ancaq yeni gələn message-ləri oxuyacaq. Default latest-dir.
none olsa valid offset olmadıqda exception verəcək. Hazırkı Kafka-da by_duration seçimi də var, hansı ki məsələn qeyd etdiymiz müddət əvvəldən başlamağa imkan verir.
Kafka message-i consumer oxuduqdan sonra silmir. Bu Kafka-nı digər queue sistemlərindən fərqləndirən əsas məsələlərdən biridir. Message topic-də retention configuration-a uyğun olaraq qalır. Eyni topic-i Fraud Service, Notification Service, Reporting Service oxuya bilər və bunların hər birinin ayrı consumer group-u və ayrı offset-i olur, biri digərinə təsir etmir.
Bunun başqa böyük üstünlüyü replay-dır. Məsələn Reporting Service-də bug tapdıq və son 2 günün transaction-larını yanlış hesablayıb. Bug-u düzəltdikdən sonra offset-i əvvələ çəkib həmin event-ləri yenidən oxuya bilərik və ya başqa group.id yaradıb topic-i yenidən consume edə bilərik.
Beləliklə, Kafka-nı production-da düzgün istifadə etmək üçün dərinliklərini bilmək vacibdir. Kafka-da bir çox configuration var amma əsas məsələ həmin parametrləri əzbərləmək deyil, hər parametrin hansı failure problemini həll etdiyini başa düşməkdir.
Oxuduğunuz üçün təşəkkürlər.
Mənbələr :
