Kafka: partition, replication, consumer group ve hata yönetimi

Bu bölümde

Kafka'yı "mesaj kuyruğu" olarak değil, partition'lara bölünmüş, replike edilen bir append-only log olarak anlayacaksın. Paralelliğin, sıralamanın ve dayanıklılığın nereden geldiğini, hangi config'in neyi garanti ettiğini ve neyi etmediğini öğreneceksin.

Mülakatta Kafka üç şekilde gelir: soru-cevap ("sıra nasıl garanti edilir", "exactly-once var mı"), hesap ("6 partition, 12 consumer, kaç tanesi çalışır") ve kod okuma (exception yutan, timeout'suz, idempotent olmayan bir @KafkaListener). Bu bölüm üçüne de hazırlar.

Konu anlatımı

Log modeli: topic, partition, offset, segment

Topic mantıksal bir isimdir; veri fiziksel olarak partition'larda durur. Her partition sıralı, sadece sona eklenen (append-only) bir log'dur. Bir kayda partition içindeki sıra numarası olan offset ile erişilir. Offset partition'a özeldir: "offset 42" ancak "order-events, partition 3, offset 42" diye anlam taşır.

Kafka'da tüketmek veriyi silmez. Kayıtlar retention süresi dolana kadar (varsayılan retention.ms 7 gün) ya da boyut sınırına gelene kadar diskte kalır. Consumer sadece "nerede kaldım" bilgisini (committed offset) tutar. Bu yüzden aynı veriyi birden fazla bağımsız sistem okuyabilir ve bir consumer offset'ini geri sararak geçmişi yeniden işleyebilir (replay). RabbitMQ gibi klasik kuyruklardan temel fark budur.

Partition diskte tek bir dosya değildir; segment'lere bölünür. Her segment bir .log dosyası ile ona eşlik eden .index (offset → dosya içi pozisyon) ve .timeindex (zaman damgası → offset) dosyalarından oluşur. Yazma her zaman aktif (son) segmente yapılır. Segment segment.bytes (varsayılan 1 GiB) ya da segment.ms sınırına gelince kapanır, yenisi açılır.

Segment'in neden var olduğu mülakatta iyi bir detaydır:

• Retention segment bazında çalışır. Kafka tek tek kayıt silmez; süresi dolan kapalı segment dosyasını komple siler. Bu, silmeyi O(1) bir dosya işlemi yapar.
• Aktif segment silinmez ve compaction'a girmez. Düşük trafikli bir topic'te retention'ın "geç" çalışıyor gibi görünmesinin sebebi budur.
• Offset ile okuma hızlıdır. Segment dosyalarının adı başlangıç offset'idir; doğru segment ikili aramayla, segment içindeki konum seyrek index ile bulunur.

Kafka'nın yüksek throughput'u da bu modelden gelir: disk'e sıralı yazma, işletim sisteminin page cache'ine güvenme, batch'ler halinde sıkıştırılmış veri ve consumer'a gönderirken zero-copy (sendfile). Kafka her mesajda fsync yapmaz; dayanıklılığı disk flush'ı yerine replication ile sağlar. Bu nokta replication bölümünde önemli olacak.

Broker, controller ve KRaft

Broker, partition replikalarını tutan ve istemcilere hizmet veren Kafka sunucusudur. Küme metadata'sını (hangi topic'ler var, hangi partition'ın leader'ı kim, ISR kimlerden oluşuyor) yöneten rol ise controller'dır.

Eskiden bu metadata Apache ZooKeeper'da tutulurdu. KRaft (Kafka Raft, KIP-500) ile metadata, Kafka'nın kendi içindeki bir Raft quorum'unda, __cluster_metadata adlı dahili bir log'da tutulur. Controller rolündeki node'lar bu log'u Raft ile replike eder; broker'lar log'u takip ederek metadata'yı öğrenir.

Sürüm bilgisi mülakatta sorulabilir:

• KRaft, Kafka 3.3'te production-ready ilan edildi.
• Kafka 4.0 (Mart 2025) ZooKeeper'sız çalışan ilk major sürümdür; ZooKeeper modu tamamen kaldırıldı. ZooKeeper modundaki bir küme 4.x'e yükseltilmeden önce KRaft'a migrate edilmek zorundadır.
• 4.0 ile istemciler ve Streams en az Java 11, broker'lar en az Java 17 ister.

KRaft'ın pratik faydası: tek bir sistemi işletirsin (ZooKeeper ensemble'ı yok), controller failover'ı metadata'yı baştan yüklemek zorunda olmadığı için hızlıdır ve partition sayısı ölçeği büyür. Lokal geliştirmede tek node'un hem broker hem controller rolünü üstlenmesi (KAFKA_PROCESS_ROLES: broker,controller) KRaft'ın combined modudur; production'da controller'lar genelde ayrı node'lardır.

Replication: leader, ISR, acks ve min.insync.replicas

Her partition'ın replication.factor kadar kopyası (replika) vardır ve bunlar farklı broker'lardadır. Replikalardan biri leader'dır; producer'lar leader'a yazar, consumer'lar varsayılan olarak leader'dan okur. Diğerleri follower'dır ve leader'dan veri çeker (fetch).

ISR (in-sync replicas), leader'a yetişmiş replikaların kümesidir. Bir follower replica.lag.time.max.ms süresince leader'a yetişemezse ISR'dan çıkarılır. Leader ölürse yeni leader sadece ISR içinden seçilir; çünkü sadece ISR'daki replikaların onaylanmış tüm veriye sahip olduğu bilinir.

High watermark, tüm ISR replikalarına yazılmış son offset'tir. Consumer'lar sadece high watermark'a kadar olan kayıtları görür. Böylece leader değişse bile consumer'ın gördüğü bir kayıt "kaybolmaz".

Producer tarafındaki acks ayarı, "yazma başarılı" demeden önce neyin bekleneceğini belirler:

Tablonun son satırı kritik: acks=all "tüm replikalar" değil, "ISR'daki tüm replikalar" demektir. ISR'da sadece leader kaldıysa acks=all tek kopyayla onay verir. Bunu engelleyen ayar min.insync.replicas'dır (topic veya broker seviyesinde, varsayılanı 1). acks=all ile birlikte kullanıldığında ISR bu sayının altına düşerse producer NotEnoughReplicas hatası alır; yani sistem veri kaybı riski yerine yazmayı reddetmeyi seçer.

Klasik production ayarı: RF=3, min.insync.replicas=2, acks=all. Bu kombinasyonla:

• Bir broker ölürse yazma devam eder (ISR 2'de kalır) ve onaylanmış hiçbir kayıt kaybolmaz.
• İki broker ölürse yazma durur (availability'den feda), ama onaylanmış veri yine kaybolmaz.

Bu garantinin bir ön koşulu daha vardır: unclean.leader.election.enable=false (varsayılan). Bu ayar true olursa ISR dışındaki geride kalmış bir replika leader olabilir ve onaylanmış kayıtlar sessizce kaybolur.

Producer: partition seçimi, sıralama, idempotence, batching

Partition seçimi. Kayıt bir key ile gönderilirse varsayılan partitioner key'in murmur2 hash'ini alır ve partition sayısına göre mod'lar: partition = toPositive(murmur2(keyBytes)) % numPartitions. Aynı key her zaman aynı partition'a gider. Key null ise Kafka 3.3+ "sticky" davranır: bir partition'a batch.size kadar veri gidene kadar orada kalır, sonra başka bir partition'a geçer. Bu, küçük batch'lerin önüne geçer.

Sıralama garantisi sadece partition içindedir. Topic genelinde global sıra yoktur. "Bir siparişin event'leri sırayla işlensin" istiyorsan key'i orderId yaparsın; böylece o siparişin tüm event'leri aynı partition'a düşer. Global sıra istiyorsan tek partition'a mahkûm olursun ve paralellik biter.

Bu formülün bir sonucu: partition sayısını artırmak key → partition eşleşmesini değiştirir. hash % 6 ile hash % 12 farklı sonuç verir. Artırma anında bir siparişin eski event'leri eski partition'da, yenileri başka partition'da olur ve iki farklı consumer bunları paralel işleyebilir; sipariş bazında sıra geçiş anında bozulur. Partition sayısı da azaltılamaz. Bu yüzden partition sayısı baştan büyümeyi hesaba katarak seçilir.

Retry ve sıra. Producer geçici hatalarda otomatik retry yapar (retries varsayılanı Integer.MAX_VALUE; asıl sınırı delivery.timeout.ms, varsayılan 120 sn koyar). Aynı bağlantıda onayı beklenmeden gönderilebilecek istek sayısı max.in.flight.requests.per.connection'dır (varsayılan 5). Idempotence kapalıyken bu değer 1'den büyükse ve bir batch fail olup retry edilirse, arkasındaki batch önce yazılabilir: sıra bozulur.

Idempotent producer. enable.idempotence=true olduğunda broker her producer'a bir producer id (PID) verir ve producer her partition için artan bir sequence number gönderir. Broker aynı PID ve sequence'ı ikinci kez görürse kaydı tekrar yazmaz; bir sequence atlanırsa kabul etmez. Sonuç:

• Retry'lar partition içinde duplicate üretmez.
• max.in.flight.requests.per.connection <= 5 olduğu sürece sıra korunur.

Varsayılanlar sürüme göre değişti ve mülakatta sorulur:

• Kafka 3.0 ile acks varsayılanı 1'den all'a döndü ve enable.idempotence varsayılan true oldu. 3.0.0 ve 3.1.0'daki bir hata yüzünden bu varsayılan fiilen ancak 3.0.1, 3.1.1 ve 3.2.0'dan itibaren uygulanır.
• Idempotence, çelişen bir config yoksa varsayılan olarak açıktır. Çelişen config varsa (ör. acks=1) ve idempotence'ı açıkça istemediysen sessizce kapanır; açıkça true dediysen ConfigException alırsın.

Batching. Producer kayıtları partition başına batch'lerde toplar. batch.size (varsayılan 16384 byte) batch'in üst sınırıdır; linger.ms batch dolmasa bile göndermeden önce ne kadar bekleneceğidir. Kafka 4.0 ile linger.ms varsayılanı 0'dan 5 ms'ye çıktı (KIP-1030). Biraz beklemek daha büyük batch, daha iyi sıkıştırma (compression.type, sıkıştırma batch bazında yapılır) ve daha az istek demektir; bedeli birkaç milisaniyelik ek gecikmedir.

send() asenkron'dur: kaydı producer'ın buffer'ına koyar ve bir Future döner. Ama tamamen bloklamaz da değildir: metadata yoksa ya da buffer doluysa max.block.ms (varsayılan 60 sn) kadar bloklayabilir. Outbox relay gibi "Kafka kapalıyken ne olur" sorusunun önemli olduğu yerlerde bu değer bilinçli seçilmelidir (bkz. Bölüm 13).

Güvenilir bir producer için Spring Boot ayarı:

spring:
  kafka:
    producer:
      acks: all
      properties:
        enable.idempotence: true
        delivery.timeout.ms: 120000
        max.block.ms: 10000
        linger.ms: 5
        compression.type: lz4

Consumer group ve partition ataması

Aynı group.id'yi kullanan consumer'lar bir consumer group oluşturur. Grup içinde her partition aynı anda tek bir consumer'a atanır; bir consumer birden fazla partition alabilir. Bu, işin paylaştırılması (queue semantiği) demektir. Farklı group.id'li gruplar ise aynı topic'in tamamını bağımsız olarak okur (pub/sub semantiği). Kargo servisi ve fatura servisi aynı order-created topic'ini ayrı gruplarla okur, her biri her event'i alır.

Buradan çıkan en önemli kural: bir grubun paralelliğinin üst sınırı partition sayısıdır. Partition'dan fazla consumer varsa fazlası boşta bekler (hot standby olarak işe yarar, ama throughput katmaz).

Spring Kafka'da concurrency = "3", aynı pod içinde 3 ayrı KafkaConsumer (3 ayrı grup üyesi, her biri kendi thread'inde) oluşturur. Toplam grup üyesi = pod sayısı × concurrency. Her üye kendi heartbeat'ini gönderir ve rebalance'a katılır; yani boşta duran üyeler de rebalance maliyetini artırır.

Atamayı grup içindeki bir üye (group leader) client-side assignor ile hesaplar (klasik protokol). Varsayılan partition.assignment.strategy listesi [RangeAssignor, CooperativeStickyAssignor]'dır ve grup, tüm üyelerin desteklediği ilk stratejiyi kullanır. Yani varsayılan ayarla grubun kullandığı strateji eager çalışan RangeAssignor'dır. Cooperative'e geçmek için listeden RangeAssignor'ı çıkarman gerekir (dokümantasyonun önerdiği yol: tek bir rolling restart ile çıkarmak).

Grubun durumunu (üyeler, atamalar, committed offset'ler) yöneten broker group coordinator'dır. Committed offset'ler __consumer_offsets adlı dahili, compacted bir topic'te tutulur.

Rebalance: eager, cooperative, static membership, KIP-848

Rebalance, partition'ların grup üyelerine yeniden dağıtılmasıdır. Tetikleyiciler: üye katılması, üye ayrılması, üyenin ölü sayılması, topic'e partition eklenmesi.

Bir üyenin "ölü" sayılmasının iki ayrı yolu vardır ve ikisi farklı problemi yakalar:

• session.timeout.ms (varsayılan 45 sn, Kafka 3.0+): Coordinator bu süre boyunca heartbeat almazsa üyeyi atar. Heartbeat'ler ayrı bir arka plan thread'inden gider (heartbeat.interval.ms, varsayılan 3 sn). Bu, process'in ya da ağın ölmesini yakalar.
• max.poll.interval.ms (varsayılan 5 dk): Uygulama iki poll() çağrısı arasında bu süreyi aşarsa consumer kendini takılmış sayar, gruptan ayrılır ve rebalance tetiklenir. Bu, heartbeat thread'i yaşarken işleme thread'inin takılmasını yakalar. Örnek: timeout'suz bir HTTP çağrısında bekleyen listener.

max.poll.records (varsayılan 500) tek poll()'un döndürebileceği kayıt sayısıdır. Basit bir hesap yapılmalı: max.poll.records × kayıt başına en kötü işleme süresi < max.poll.interval.ms. Kayıt başına 1 saniye süren bir işte 500 kayıt 500 saniye eder ve 300 saniyelik sınırı aşar.

Eager rebalance (Range, RoundRobin, eski Sticky): Rebalance başladığında tüm üyeler tüm partition'larını bırakır, yeni atama gelene kadar grup durur (stop-the-world). 100 üyeli bir grupta tek bir pod restart'ı tüm tüketimi saniyelerce durdurur.

Cooperative (incremental) rebalance (CooperativeStickyAssignor, Kafka 2.4+): Üyeler sadece başka birine taşınması gereken partition'ları bırakır; diğerlerini işlemeye devam eder. Bedeli, bir rebalance'ın iki tur sürebilmesidir.

Static membership (group.instance.id, Kafka 2.3+): Her consumer'a kalıcı bir kimlik verilir (Kubernetes'te StatefulSet pod adı iyi bir adaydır). Statik üye restart olup session.timeout.ms dolmadan geri gelirse rebalance olmaz, aynı partition'ları geri alır. Rolling deploy'larda rebalance fırtınasını önler. Bedeli: gerçekten ölen bir üyenin partition'ları ancak session timeout dolunca başkasına geçer; bu süre boyunca o partition'larda lag birikir.

KIP-848, yeni consumer group protokolü: Kafka 4.0'da GA oldu. Atama hesabı client'tan broker'daki coordinator'a taşınır; rebalance'lar incremental'dır ve tüm grubu senkronize eden bir bariyer yoktur. Sunucu tarafında 4.0'a geçişle otomatik etkinleşir, ama istemcinin group.protocol=consumer ile açıkça seçmesi gerekir; consumer'da group.protocol varsayılanı hâlâ classic'tir. Yeni protokolde session.timeout.ms ve heartbeat.interval.ms istemcide değil broker'da (group.consumer.*) yönetilir.

Offset commit ve teslimat semantiği

Committed offset "işlenen son kayıt" değil, **"okunacak bir sonraki kayıt"**tır: offset 41'i işlediysen 42'yi commit edersin. Consumer yeniden başladığında veya partition başkasına geçtiğinde okuma committed offset'ten devam eder. Grup için hiç commit yoksa auto.offset.reset devreye girer (varsayılan latest; yeni bir grup topic'teki eski mesajları görmez, earliest ile baştan başlar).

Teslimat semantiği, işleme ile commit'in sırasına bağlıdır:

Auto commit'in gerçek tehlikesi. enable.auto.commit=true (Kafka client varsayılanı) olduğunda commit, auto.commit.interval.ms (5 sn) aralıklarla, poll() çağrısı içinde, bir önceki poll'un döndürdüğü offset'ler için yapılır. Kayıtları poll döngüsü içinde senkron işliyorsan bu aslında at-least-once davranır. Tehlike, kayıtları başka bir thread havuzuna devredip hemen tekrar poll() çağırdığında ortaya çıkar: işlenmemiş kayıtların offset'i commit edilir, pod ölürse bu kayıtlar kaybolur. Ayrıca commit zamanlaması senin kontrolünde değildir.

Spring Kafka bu yüzden 2.3'ten beri enable.auto.commit'i (açıkça set etmediysen) false yapar ve commit'i container üstlenir. Varsayılan AckMode.BATCH: bir poll()'un döndürdüğü tüm kayıtlar listener'dan başarıyla geçince offset'ler commit edilir. RECORD her kayıttan sonra commit eder; MANUAL ve MANUAL_IMMEDIATE commit'i Acknowledgment.acknowledge() çağrısına bağlar.

Spring Kafka'da listener exception fırlatmadan dönerse kayıt "başarılı" sayılır ve offset commit edilir. Exception'ı catch edip loglamak, mesajı sessizce çöpe atmak demektir. q13'ün birinci problemi tam olarak budur.

Exactly-once: ne garanti eder, ne etmez

Kafka'nın exactly-once semantiği (EOS) üç parçadan oluşur:

1. Idempotent producer: Producer retry'larında duplicate yok.
2. Transactions: transactional.id verilmiş bir producer, birden fazla partition'a yazmayı ve consumer offset commit'ini (sendOffsetsToTransaction) tek bir atomik transaction içinde yapar. Aynı transactional.id ile yeni bir producer başladığında epoch artar ve eski "zombie" producer'ın yazmaları reddedilir (fencing).
3. isolation.level=read_committed: Consumer sadece commit edilmiş transaction'ların kayıtlarını görür; abort edilen transaction'ların kayıtlarını atlar. Varsayılan read_uncommitted'tır; bu ayarı unutursan transaction'ın abort edilen kayıtlarını da okursun.

Bu üçlü birlikte "Kafka'dan oku, işle, Kafka'ya yaz" (read-process-write) döngüsünü exactly-once yapar. Kafka Streams'te processing.guarantee=exactly_once_v2 bunu tek ayarla verir.

Sınır net: dış sistem transaction'ın parçası değildir. Consumer PostgreSQL'e yazıyor, bir HTTP API çağırıyor veya e-posta gönderiyorsa Kafka transaction'ı bu yan etkiyi geri alamaz. DB commit oldu, Kafka offset commit'i olmadan pod öldü: mesaj tekrar gelir ve DB'ye ikinci kez yazılır. Burada çözüm Kafka ayarı değil, uygulama tasarımıdır:

• Idempotent consumer: İşlenen mesaj id'sini iş verisiyle aynı DB transaction'ında bir processed_event tablosuna yaz; unique constraint ikinci denemeyi yakalasın (ayrıntı için bkz. Bölüm 13).
• Offset'i DB'de tutmak: Offset'i iş verisiyle aynı transaction'da DB'ye yazıp başlangıçta seek() ile oradan devam etmek. Güçlü ama Spring Kafka'nın commit modelinin dışına çıkarsın.
• Doğal idempotency: Upsert, UNIQUE(order_id), "versiyonu benimkinden büyükse güncelle" gibi koşullu yazmalar.

Takip sorusunun cevabı: Hayır. İki ayrı kaynağın transaction'ları ardışık commit edilir (best-effort 1PC); biri commit olup diğeri fail edebilir. Bu yüzden DB yan etkisi olan consumer'da yine idempotency gerekir.

Spring Kafka'da hata yönetimi

Listener exception fırlattığında kaydın ne olacağına container'a bağlı CommonErrorHandler karar verir. Spring Boot, context'te tek bir CommonErrorHandler bean'i varsa onu otomatik olarak listener container factory'ye bağlar.

DefaultErrorHandler (blocking retry). Varsayılan davranış:

• Backoff FixedBackOff(0L, 9): bekleme olmadan 9 retry, yani toplam 10 deneme. Sonra kayıt (varsayılan recoverer ile) loglanır ve atlanır.
• Retry, kalan kayıtlara seek yapılarak gerçekleşir: hatalı kayıt ve arkasındakiler bir sonraki poll()'da tekrar gelir. Aynı partition'daki sonraki kayıtlar bekler; sıra korunur, ama partition tıkanır.
• Backoff süresi boyunca consumer thread'i uyur. Tek bir backoff aralığı + işleme süresi max.poll.interval.ms'i aşarsa rebalance olur. Uzun gecikmeler için ContainerPausingBackOffHandler container'ı pause ederek bekler.

Retry edilmeyen (fatal) exception'lar. DefaultErrorHandler bazı exception'ları varsayılan olarak retry etmez ve ilk hatada recoverer'ı çağırır: DeserializationException, MessageConversionException, ConversionException, MethodArgumentResolutionException, NoSuchMethodException, ClassCastException. Kendi iş kuralı hatalarını addNotRetryableExceptions(...) ile ekleyebilirsin. Ayrıca DefaultErrorHandler sadece RuntimeException'ları işler; Error alt sınıfları (ör. OutOfMemoryError) error handler'ı atlar.

DeadLetterPublishingRecoverer (DLT). Denemeler bitince kaydı bir dead-letter topic'e yayınlar ve orijinal topic, partition, offset, exception sınıfı, mesajı ve stack trace'i header olarak ekler. Varsayılan hedef orijinal kaydın partition'ı ile aynı numaralı partition'dır; bu yüzden DLT'nin en az kaynak topic kadar partition'ı olmalıdır.

@RetryableTopic (non-blocking retry). Hatalı kayıt gecikme bilgisiyle ayrı retry topic'lerine taşınır, ana partition ilerlemeye devam eder. Retry topic consumer'ı, zamanı gelmemiş kayıt için o partition'ı pause eder; zamanı gelince tekrar dener; denemeler bitince kayıt DLT'ye gider. Varsayılanlar: 3 deneme (ilk deneme dahil), 1000 ms sabit backoff, sonekler -retry ve -dlt. Üssel backoff ile main-topic-retry-1000, main-topic-retry-2000 gibi topic'ler oluşur (sabit gecikmede tek veya index'li retry topic kullanımı sürüme göre değişir).

Bedeller:

• Sıra garantisi kaybolur. Hatalı kayıt retry topic'te beklerken aynı key'in sonraki kayıtları ana topic'te işlenir. "Sipariş iptal" event'i "sipariş oluştu" event'inden önce işlenebilir.
• Batch listener'larla ve container transaction'larıyla birlikte kullanılamaz.
• Topic sayısı artar; her retry topic izlenmelidir.

Hangisini ne zaman seçersin:

Son satır senior bir detaydır: e-fatura sağlayıcısı 10 dakikalığına çöktüyse o 10 dakikadaki her mesajı DLT'ye atmak, sonra hepsini elle replay etmek demektir. Daha iyisi, circuit breaker açıkken listener container'ı pause etmek (KafkaListenerEndpointRegistry üzerinden) ve lag'in birikmesine izin vermektir. Kafka zaten bir buffer'dır.

Poison pill ve deserialization. Her denemede aynı şekilde fail eden mesaja poison pill denir. En tehlikeli türü deserialization hatasıdır: Kafka client'ı poll() içinde deserialize eder; deserializer exception fırlatırsa kayıt listener'a hiç ulaşmaz ve consumer aynı offset'te sonsuza kadar takılabilir. Çözüm ErrorHandlingDeserializer: gerçek deserializer'ı sarar, hata olursa exception'ı kaydın header'ına koyar ve değeri null olarak teslim eder. Container bunu görür, DeserializationException fatal olduğu için retry etmeden recoverer'a verir; DeadLetterPublishingRecoverer DLT'ye orijinal byte[] değeri yazar.

spring:
  kafka:
    consumer:
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
      properties:
        spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
        spring.json.trusted.packages: "com.example.order.events"

Tam bir hata yönetimi kurulumu:

@Configuration
public class KafkaErrorHandlingConfig {

    @Bean
    public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<Object, Object> template) {
        // DLT adını açıkça veriyoruz: sürümler arası varsayılan farkından etkilenmez.
        var recoverer = new DeadLetterPublishingRecoverer(template,
                (record, ex) -> new TopicPartition(record.topic() + "-dlt", record.partition()));

        // 1 sn, 2 sn, 4 sn, 8 sn: toplam 5 deneme. Her aralık max.poll.interval.ms'ten çok küçük.
        var backOff = new ExponentialBackOffWithMaxRetries(4);
        backOff.setInitialInterval(1_000L);
        backOff.setMultiplier(2.0);
        backOff.setMaxInterval(10_000L);

        var handler = new DefaultErrorHandler(recoverer, backOff);
        // İş kuralı ihlali tekrar denemekle düzelmez: doğrudan DLT.
        handler.addNotRetryableExceptions(InvalidOrderEventException.class);
        return handler;
    }
}

DLT bir çöp kutusu değil, bir iş kuyruğudur: DLT'ye düşen her mesaj için alarm, sahiplik, inceleme ve düzeltildikten sonra replay (DLT'den okuyup ana topic'e geri yazan bir araç veya endpoint) süreci olmalıdır.

Lag, izleme ve ölçekleme

Consumer lag, bir partition'ın son offset'i (log end offset) ile grubun committed offset'i arasındaki farktır. Lag sürekli artıyorsa consumer'lar üretim hızına yetişemiyor demektir.

Lag'i görmenin yolları:

• kafka-consumer-groups.sh --bootstrap-server ... --describe --group invoice-service: partition başına CURRENT-OFFSET, LOG-END-OFFSET, LAG ve hangi consumer'a atandığını gösterir.
• Consumer client metriği records-lag-max (Micrometer ile Spring Boot'ta otomatik yayınlanır).
• Kümeyi dışarıdan izleyen exporter'lar ve yönetilen servislerin panelleri.

İzlerken dikkat edilecekler:

• Kayıt sayısı yerine zaman cinsinden lag daha anlamlıdır: "10.000 mesaj geride" trafiğe göre 1 saniye de olabilir 1 saat de. "En eski işlenmemiş mesaj 4 dakikalık" iş tarafına doğrudan anlam taşır.
• Toplam lag değil, partition bazında lag'e bak. Tek bir partition'da büyüyen lag hot key (tek bir satıcının tüm trafiği) veya takılmış bir consumer'a (poison pill) işaret eder.
• DLT'ye yazılan mesaj sayısı ve rebalance sayısı da ayrı metrik olarak izlenmelidir.

Ölçekleme. Lag'e göre otomatik ölçekleme (Kubernetes'te KEDA'nın Kafka scaler'ı) yaygındır. Kural: maksimum replika × concurrency, partition sayısını geçmemeli; geçerse eklenen pod'lar boşta kalır ama her ekleme bir rebalance tetikler. Asıl paralellik ihtiyacı partition sayısıyla planlanır.

Quiz'deki sorular: çözüm ve nedenleri

q12: 6 partition, 4 pod, concurrency=3

Senaryo. order-events topic'i 6 partition, RF=3. Kargo servisi 4 pod, hepsi aynı groupId, listener'da concurrency = "3". Producer'lar key olarak orderId gönderiyor. Hangi ifadeler doğru?

Doğru cevaplar: a, b, d, f.

a) "Toplam 12 consumer thread var, ama aynı anda sadece 6'sı mesaj işler." Doğru. 4 pod × 3 concurrency = 12 grup üyesi. Grup içinde bir partition aynı anda tek üyeye atanır; 6 partition'ı en fazla 6 üye alabilir, diğer 6'sı hiç partition almaz. Hangi 6 üyenin iş alacağı assignor'a ve üye kimliklerinin sıralamasına bağlıdır; bir pod'un hiç partition almaması da, bir pod'un iki partition alması da mümkündür. Boştaki 6 üye boşuna değildir: bir üye ölürse onun partition'ını devralırlar. Ama throughput katmazlar ve her rebalance'a katılarak süresini uzatırlar.

b) "Key = orderId olduğu için aynı siparişin event'leri aynı partition'a gider ve o partition içinde sırayla işlenir." Doğru. murmur2(key) % 6 aynı key için hep aynı sonucu verir. Partition tek bir consumer thread'inde sırayla işlendiği için o siparişin event'leri sırayla işlenir. Bu garantinin iki ön koşulu var: producer'da idempotence açık (ya da max.in.flight=1) olmalı ki retry sırayı bozmasın; consumer'da da kayıtlar başka bir thread havuzuna dağıtılmamalı.

c) "Partition sayısını 6'dan 12'ye çıkarmak mevcut key'lerin partition eşleşmesini değiştirmez." Yanlış. Formül hash % N olduğu için N değişince key'lerin çoğu başka partition'a düşer ve geçiş anında aynı siparişin event'leri iki partition'a dağılır.

d) "acks=all + min.insync.replicas=2 + RF=3 ile tek broker kaybında onaylanmış mesaj kaybolmaz." Doğru. Onay en az 2 ISR replikasına yazıldıktan sonra gelir. Biri ölürse veriye sahip diğer ISR replikası leader olur. Ön koşul: unclean.leader.election.enable=false (varsayılan).

e) "Kafka'nın exactly-once desteği, consumer'ın PostgreSQL'e yazdığı veride de duplicate olmayacağını garanti eder." Yanlış. EOS sadece Kafka'dan okuyup Kafka'ya yazan akışı kapsar; PostgreSQL yazması Kafka transaction'ının parçası değildir.

f) "max.poll.interval.ms aşılırsa consumer gruptan atılır ve rebalance tetiklenir." Doğru. Heartbeat thread'i yaşasa bile iki poll() arası bu süreyi aşan consumer takılmış sayılır. Küçük bir nüans: group.instance.id kullanan statik bir üyede partition'lar hemen yeniden dağıtılmaz; consumer heartbeat'i keser ve atama session timeout dolunca değişir.

Arkasındaki kavram. Partition, Kafka'da hem paralelliğin hem sıralamanın birimidir. Bu tek cümleden üç production kuralı çıkar: consumer sayısını partition'a göre planla, sıralama gereken her şeyi aynı key ile gönder, partition sayısını baştan büyümeye göre seç.

q13: Fatura consumer'ında kaç problem var?

Senaryo. order-created topic'i 6 partition. E-fatura sağlayıcısı kampanya günlerinde 20-40 saniye cevap vermiyor.

@KafkaListener(topics = "order-created", groupId = "invoice-service", concurrency = "12")
public void onOrderCreated(String payload) {
    try {
        OrderCreated event = objectMapper.readValue(payload, OrderCreated.class);
        Invoice invoice = invoiceClient.create(event);   // harici e-fatura API, timeout ayarı yok
        invoiceRepository.save(invoice);
    } catch (Exception e) {
        log.error("Fatura oluşturulamadı", e);
    }
}

Problem 1: Exception yutuluyor, fatura sessizce kayboluyor. Metot exception fırlatmadan döndüğü için container kaydı başarılı sayar ve AckMode.BATCH ile offset'i commit eder. Sağlayıcı 503 döndüyse o siparişin faturası hiç kesilmez ve kimsenin haberi olmaz; tek iz bir log satırıdır. Çözüm: catch'i kaldır, exception'ı container'a bırak. DefaultErrorHandler + backoff + DeadLetterPublishingRecoverer kur. DLT için alarm ve replay süreci tanımla.

Problem 2: HTTP çağrısında timeout yok. Varsayılan JDK HTTP istemcilerinde read timeout sonsuzdur. Hesap: max.poll.records 500, çağrı başına 40 sn. Sekizinci kayıtta 8 × 40 = 320 sn ile max.poll.interval.ms (300 sn) aşılır. Consumer gruptan atılır, rebalance olur; BATCH modunda o poll'un hiçbir offset'i commit edilmediği için işlenmiş 7 kayıt partition'ı devralan consumer'da tekrar işlenir. Yani timeout eksikliği doğrudan duplicate faturaya dönüşür. Çözüm: Connect timeout (ör. 1-2 sn) ve read timeout (sağlayıcının normal p99'unun biraz üstü, ör. 5 sn), circuit breaker, gerekirse max.poll.records'u düşürmek. (Timeout seçimi için bkz. Bölüm 14.)

Problem 3: Consumer idempotent değil. At-least-once teslimatta aynı mesaj rebalance, retry veya pod ölümüyle birden fazla kez gelir. Her gelişte yeni bir e-fatura kesilir. E-fatura yasal bir belge olduğu için ikinci fatura iptal süreci ve muhasebe sorunu demektir. Çözüm: order_id üzerinde unique constraint, sağlayıcıya orderId tabanlı idempotency key (sağlayıcı destekliyorsa), işlenmiş event tablosu.

Problem 4: concurrency = 12, partition = 6. Pod başına 12 consumer oluşur; birden fazla pod varsa boşta kalan sayısı daha da artar. 6'dan fazlası iş almaz. Çözüm: Concurrency'yi partition sayısına göre planla (pod sayısı × concurrency ≤ partition).

Problem 5: Harici çağrı ile DB kaydı atomik değil. Fatura sağlayıcıda kesildi, invoiceRepository.save fail oldu (DB anlık erişilemez). Retry'da sağlayıcıya tekrar gidilir ve ikinci fatura kesilir. Harici sistem DB transaction'ının parçası olamaz. Çözüm: Önce DB'ye PENDING durumunda bir fatura isteği yaz (unique order_id), sonra sağlayıcıya o kaydın id'sini idempotency key olarak göndererek git, dönünce kaydı ISSUED yap. Retry'da PENDING kayıt bulunur ve aynı key ile tekrar denenir.

Problem 6: Parse hatası (poison pill) retry ediliyor (exception yutulmasa bile). Bozuk JSON 10 kez denense de düzelmez. Çözüm: ErrorHandlingDeserializer ile JSON'u container'a parse ettir (deserialization hataları zaten fatal sayılır) ya da parse hatasını not-retryable bir exception'a sarıp addNotRetryableExceptions ile doğrudan DLT'ye gönder.

Bonus (senior): Yavaş bir sağlayıcıyı consumer thread'inde beklemek. 20-40 saniyelik çağrılar consumer thread'ini, dolayısıyla partition'ı kilitler. Daha iyi bir tasarım işi ikiye böler: consumer sadece PENDING fatura isteğini DB'ye yazar (milisaniyeler, idempotent), ayrı bir worker (rate limit + circuit breaker ile) sağlayıcıyı çağırır. Kafka tüketimi sağlayıcının sağlığından bağımsız hale gelir.

Daha iyi bir çözüm:

@Component
public class InvoiceRequestedConsumer {

    private final InvoiceRequestService invoiceRequestService;

    public InvoiceRequestedConsumer(InvoiceRequestService invoiceRequestService) {
        this.invoiceRequestService = invoiceRequestService;
    }

    // concurrency: 6 partition / 2 pod = 3. JSON'u ErrorHandlingDeserializer + JsonDeserializer çözer.
    @KafkaListener(topics = "order-created", groupId = "invoice-service", concurrency = "3")
    public void onOrderCreated(OrderCreated event) {
        // try/catch YOK: exception DefaultErrorHandler'a gider (backoff -> DLT).
        invoiceRequestService.registerPending(event);
    }
}

@Service
public class InvoiceRequestService {

    private final InvoiceRequestRepository repository;

    public InvoiceRequestService(InvoiceRequestRepository repository) {
        this.repository = repository;
    }

    @Transactional
    public void registerPending(OrderCreated event) {
        // INSERT ... ON CONFLICT (order_id) DO NOTHING: aynı event 5 kez gelse de tek kayıt.
        repository.insertPendingIfAbsent(UUID.randomUUID(), event.orderId(), event.totalAmount());
    }
}

Faturayı kesen worker (@Scheduled ile PENDING kayıtları FOR UPDATE SKIP LOCKED ile alan, timeout ve circuit breaker'lı bir istemciyle sağlayıcıya invoiceRequestId'yi idempotency key olarak gönderen, sonucu ISSUED/FAILED olarak işleyen bir bileşen) Bölüm 13'teki outbox relay ile aynı kalıptadır.

Üç seviyede cevap

Soru: "Kafka'da mesaj sırası nasıl garanti edilir?"

Soru: "Kafka'da bir mesajın iki kez işlenmesini nasıl önlersin? Kafka exactly-once değil mi?"

Soru: "Consumer'da bir mesaj sürekli hata veriyor. Ne olur, ne yaparsın?"

Soru: "Kafka'da mesaj kaybını nasıl önlersin?"

Sorulmadı ama sorulabilir

Compacted topic

cleanup.policy=compact olan bir topic'te Kafka, her key için en az son değeri saklar; aynı key'in eski değerlerini arka planda (log cleaner) siler. Değeri null olan bir kayıt (tombstone) o key'in silinmesini ifade eder; tombstone'lar da delete.retention.ms (varsayılan 1 gün) sonra temizlenir. Aktif segment compaction'a girmez, bu yüzden compaction "anında" değildir; okuyan taraf bir key'in birden fazla değerini görebilir.

Kullanım alanları: bir varlığın son durumunu taşıyan "changelog" topic'leri (ör. ürün fiyatı, satıcı ayarları), Kafka Streams state store yedekleri, CDC snapshot'ları. Kafka'nın kendi __consumer_offsets topic'i de compacted'dır. cleanup.policy=compact,delete ile hem compaction hem süreye bağlı silme yapılabilir.

Mülakat özeti: "Compacted topic her key'in son değerini saklayan bir changelog'dur; yeni bir servis topic'i baştan okuyarak tüm varlıkların güncel halini yükleyebilir."

Kafka vs RabbitMQ

Güncel notlar: RabbitMQ 4.0 (Eylül 2024) klasik kuyruk mirroring'ini tamamen kaldırdı; replike kuyruk için quorum queue kullanılır. Kafka tarafında ise share groups (Queues for Kafka, KIP-932) partition sayısından bağımsız, mesaj başına ack'li kuyruk semantiği getirir: 4.0'da early access, 4.1'de preview, 4.2'de production-ready ilan edildi.

Mülakat özeti: "Olay akışını birden çok tüketici bağımsız okuyacak, replay ve yüksek throughput gerekiyorsa Kafka; mesaj başına ack, routing ve iş kuyruğu semantiği gerekiyorsa RabbitMQ."

Partition sayısı nasıl seçilir?

Kaba hesap: partition ≥ max(hedef throughput / tek consumer throughput, hedef throughput / tek partition producer throughput). Saniyede 30.000 mesaj, bir consumer thread'i saniyede 2.000 mesaj işliyorsa en az 15 partition gerekir; büyüme payıyla 24 veya 30 seçilir. Ayrıca partition sayısı consumer instance sayısının katları olacak şekilde seçilirse dağılım dengeli olur.

Fazla partition'ın bedelleri: her partition replika başına dosya ve bellek tüketir, producer'da partition başına batch tutulur (daha küçük batch'ler), failover'da daha çok leader değişimi olur, rebalance'lar uzar. Sayı sonradan artırılabilir ama key eşleşmesi değişir; azaltılamaz.

Mülakat özeti: "Partition sayısı paralellik tavanıdır; throughput'a ve büyümeye göre seçerim, sonradan artırmanın sıralama bedelini bilirim."

Kafka neden hızlı?

• Sıralı I/O: Sadece sona ekleme; disk ve SSD'de rastgele erişimden çok daha hızlı.
• Page cache: Kafka kendi cache'ini tutmaz, işletim sisteminin page cache'ini kullanır. Yeni yazılan veri çoğu zaman consumer'a doğrudan bellekten gider.
• Zero-copy: Disk'teki veri, kullanıcı alanına kopyalanmadan sendfile ile soket'e aktarılır (TLS açıkken bu avantaj azalır).
• Batching ve sıkıştırma: Producer batch'leri sıkıştırır, broker batch'i açmadan saklar, consumer açar.
• Basit broker: Mesaj başına ack/redelivery durumu tutulmaz; consumer sadece bir offset taşır.

Mülakat özeti: "Kafka'nın hızı sıralı yazma, page cache, zero-copy ve uçtan uca batching'den gelir; broker mesaj başına durum tutmaz."

Transactional producer'ın Spring tarafı

Spring Kafka'da spring.kafka.producer.transaction-id-prefix verildiğinde KafkaTemplate transactional olur ve KafkaTransactionManager oluşur. Listener container bir transaction manager ile yapılandırılırsa offset commit'i ve listener içindeki send()'ler aynı Kafka transaction'ında yapılır. Downstream consumer'ların isolation.level=read_committed kullanması gerekir.

Bu, Kafka → Kafka akışında güzel çalışır. DB ile birleştirmek için tek gerçek atomik yol yoktur; DB transaction'ı ile Kafka transaction'ı ardışık commit edilir. Bu yüzden DB'ye yazan servislerde Kafka transaction'ı yerine outbox tercih edilir (bkz. Bölüm 13).

Mülakat özeti: "Kafka transaction'ı Kafka içindeki yazmaları ve offset'i atomik yapar; DB ile birleştirmek istiyorsam outbox kullanırım."

Consumer'ı Kubernetes'te çalıştırmak

• Graceful shutdown: Pod kapanırken container durdurulur, son commit yapılır, consumer gruptan ayrılır. terminationGracePeriodSeconds, işlenen en uzun kaydın süresinden uzun olmalıdır.
• Static membership: group.instance.id'yi pod adından türetmek rolling deploy'da rebalance'ı azaltır.
• Liveness probe: Liveness'ı Kafka bağlantısına bağlama; broker'da bir sorun olduğunda tüm pod'ların restart döngüsüne girmesi rebalance fırtınası yaratır (bkz. Bölüm 15).
• Autoscaling: KEDA ile lag'e göre ölçekle, maksimum replika partition sayısını geçmesin.

Mülakat özeti: "Consumer pod'larında graceful shutdown, static membership ve partition'a göre sınırlanmış autoscaling kullanırım; liveness'ı Kafka'ya bağlamam."

Mesaj boyutu ve büyük payload

Broker'ın varsayılan maksimum mesaj boyutu yaklaşık 1 MB'tır (message.max.bytes); producer'da max.request.size, consumer'da max.partition.fetch.bytes ile birlikte ayarlanır. Büyük payload'ları (PDF fatura, görsel) Kafka'ya koymak yerine nesne depolamaya yazıp event'te sadece referansını (claim-check pattern) taşımak tercih edilir.

Mülakat özeti: "Kafka büyük blob'lar için değil; büyük veriyi object storage'a koyar, event'te referansını taşırım."

Sık yapılan hatalar

• Exception'ı listener'da yakalayıp loglamak → Offset commit edilir, mesaj kaybolur → Exception'ı fırlat, DefaultErrorHandler + DLT kur.
• Partition sayısından fazla consumer koyup ölçeklendiğini sanmak → Fazla consumer boşta kalır, sadece rebalance maliyeti artar → Paralellik için partition sayısını planla.
• acks=all'u "tüm replikalar" sanmak → ISR tek replikaya düşerse tek kopyayla onay verir → min.insync.replicas=2 ile birlikte kullan.
• Kafka EOS'un DB yazılarını da kapsadığını düşünmek → DB Kafka transaction'ının parçası değildir → At-least-once + idempotent consumer.
• Global sıra beklemek → Sıra sadece partition içinde → Sıra gereken varlığı key yap.
• Partition sayısını sıralama etkisini düşünmeden artırmak → Key eşleşmesi değişir, geçiş anında sıra bozulur → Artırmayı migrasyon olarak planla veya yeni topic aç.
• Listener'da timeout'suz harici çağrı → max.poll.interval.ms aşılır, rebalance ve duplicate → Timeout + circuit breaker, gerekirse işi ayrı worker'a taşı.
• Kayıtları async thread havuzuna verip offset'i hemen commit etmek → İşlenmemiş mesajlar kaybolur, sıra bozulur → İşleme bitmeden commit etme; paralellik için partition kullan.
• Sıra kritik bir akışta @RetryableTopic kullanmak → Hatalı mesaj beklerken arkasındakiler işlenir → Blocking retry + DLT.
• DLT adını ezbere .DLT yazmak → Spring Kafka 3.3+ -dlt kullanır, alarm boşa kurulur → Resolver ile adı açıkça ver.
• send() sonucunu kontrol etmemek → Asenkron hata görülmez → Callback/get() veya outbox.
• Deserialization hatasına hazırlıksız olmak → Consumer aynı offset'te takılır → ErrorHandlingDeserializer.

Hızlı tekrar

Kaynaklar

• Apache Kafka Documentation: Tasarım (log, replication, teslimat semantiği) bölümleri ve tüm config'lerin varsayılanları için birincil kaynak.
• Apache Kafka 4.0.0 Release Announcement: ZooKeeper'ın kaldırılması, KIP-848 GA ve Queues for Kafka early access.
• KIP-848: The Next Generation of the Consumer Rebalance Protocol: Yeni rebalance protokolünün neden ve nasıl tasarlandığı.
• Transactions in Apache Kafka (Confluent): transactional.id, zombie fencing ve read_committed mekaniği; EOS'un dış sistemleri neden kapsamadığı.
• Spring for Apache Kafka: Handling Exceptions: DefaultErrorHandler, fatal exception listesi, DeadLetterPublishingRecoverer, ErrorHandlingDeserializer.
• Spring for Apache Kafka: Non-Blocking Retries: @RetryableTopic'in çalışma şekli, topic isimlendirme ve sınırlamaları.
• Spring for Apache Kafka: Message Listener Containers: AckMode'lar ve offset commit davranışı.
• Kafka: The Definitive Guide, 2. baskı (Shapira, Palino, Sivaram, Petty; O'Reilly): Replication, consumer group ve güvenilir teslimat bölümleri mülakat için yeterli derinlikte.
• Designing Data-Intensive Applications (Martin Kleppmann), Bölüm 11: Log tabanlı mesajlaşmanın klasik kuyruklarla karşılaştırması.