Kafka — Partition, Consumer Group ve Lag

Kafka’yı bir kuyruk sanmak çok doğal ama yanlış. Kafka, sayfaları hiç yırtılmayan ve birkaç bölüme ayrılmış bir kayıt defteridir: mesaj okununca değil, saklama süresi dolunca silinir.

Partition: her şeyin merkezi

Bir topic, N adet partition?Terimin tanımını açpartitionBir Kafka topic'inin, mesajların sırayla eklendiği ve ayrı ayrı tüketilebilen bölümlerinden biri.Hem paralelliğin hem sıralamanın birimi budur: paralellik partition sayısıyla sınırlı, sıralama yalnızca partition içinde garanti. İkisi aynı şeye bağlı olduğu için birbiriyle yarışır.Sözlükte gör →’dan oluşur. Her partition bağımsız, sıralı, append-only bir log’dur ve her mesajın o log içinde bir offset’i vardır.

Üç kritik sonuç:

• Sıralama yalnızca partition içindedir. Topic genelinde sıralama diye bir şey yoktur.
• Paralellik tavanı partition sayısıdır. Bir consumer group’ta bir partition’ı aynı anda yalnızca bir consumer okuyabilir.
• Partition sayısını artırmak kolay, azaltmak imkânsızdır. Ve artırmak, anahtarların dağılımını bozar.

Kendin gör

Varsayılan ayarda 4 partition, 2 consumer var ve üretim tüketimden hızlı — lag?Terimin tanımını açlagÜretilen ile tüketilen offset arasındaki fark. Artıyor ve hata yoksa sorun hız farkındadır — ve paralelliğin tavanı partition sayısıdır.Sözlükte gör → büyüyor.

Sırayla dene:

• Consumer ekle’ye iki kez bas. Önce rebalance?Terimin tanımını açrebalanceConsumer group'a üye girip çıktığında partition'ların yeniden dağıtılması. Klasik biçiminde tüm grup kısa süre durur — kapasite eklemenin görünmeyen bedeli budur.Sözlükte gör → yüzünden lag artıyor, sonra kapasite yettiği için eriyor. Kapasite eklemenin kısa vadeli bedeli budur.
• Consumer sayısını 6 yap, partition 4 kalsın. İki consumer boşta etiketiyle duruyor. Hiçbir katkıları yok, sadece grup üyesi olarak rebalance’a katılıyorlar.
• Trafik patlaması’na bas. Lag birden tırmanıyor; kapasite üretimin üstündeyse eriyor, altındaysa hiç kapanmıyor.
• Consumer düşür. Düşen consumer’ın partition’ları sahipsiz kalıyor, rebalance başlıyor, o süre boyunca kimse iş yapmıyor.
• Stratejiyi round-robin yap. 4 partition / 3 consumer dağılımı değişiyor — range bir consumer’a iki partition verirken round-robin daha düzgün dağıtıyor.

Birikme basit bir hesaptır

toplam kapasite = consumer sayısı × consumer başına hız

Kapasite üretimden küçükse lag doğrusal olarak büyür. Bu bir hata değil, matematiktir.

Rebalance — kapasite eklemenin bedeli. Gruba bir consumer katıldığında veya bir consumer düştüğünde Kafka partition’ları yeniden dağıtır. Klasik (eager) protokolde bu stop-the-world’dür: tüm consumer’lar partition’larını bırakır, yeni atama yapılır, sonra herkes devam eder.

Bu sırada üreticiler durmaz. Lag tırmanır.

consumer eklendi
   │
   ├─ tüm consumer'lar durdu        ← lag tırmanmaya başlar
   ├─ koordinatör yeni atamayı hesapladı
   └─ consumer'lar yeni partition'larıyla devam ediyor

Kafka 2.4 ile gelen cooperative-sticky protokolü bunu düzeltir: yalnızca gerçekten el değiştiren partition’lar durur, geri kalan consumer’lar çalışmaya devam eder.

1
props.put(
2
    ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
3
    CooperativeStickyAssignor.class.getName()   // eager rebalance'ın duraklamasını kaldırır
4
);

Anahtar ve sıralama

Mesajın anahtarı varsa partition hash(key) % partitionCount ile seçilir. Aynı anahtar her zaman aynı partition’a gider, dolayısıyla o anahtarın mesajları sıralı işlenir.

1
// Bir siparişin olayları hep aynı partition'a düşsün diye anahtar = orderId
2
producer.send(new ProducerRecord<>("orders", order.id(), event));

Anahtar yoksa mesajlar partition’lara dağıtılır ve sıralama garantisi kalmaz.

Offset commit — teslimat garantisini belirleyen yer.

Pratikte doğru cevap neredeyse her zaman at-least-once + idempotent tüketici’dir. exactly-once yalnızca Kafka’dan okuyup Kafka’ya yazan akışlarda mümkündür; araya bir veritabanı girince yine idempotency’ye dönersin.

Aşağıda üçü bir arada: olaylar hesap numarasıyla anahtarlanıyor, offset iş bitince commit ediliyor ve tekrar gelen olay bakiyeyi ikinci kez değiştirmiyor.

1
spring:
2
  kafka:
3
    consumer:
4
      group-id: card-balance-projection
5
      enable-auto-commit: false          # commit only after the balance is written
6
      auto-offset-reset: earliest
7
      max-poll-records: 100              # keep one poll well inside max.poll.interval.ms
8
      isolation-level: read_committed
9
      properties:
10
        partition.assignment.strategy: org.apache.kafka.clients.consumer.CooperativeStickyAssignor
11
    listener:
12
      ack-mode: record                   # at-least-once: commit each record once it is processed

1
@Component
2
class CardTransactionPublisher {
3

4
    private final KafkaTemplate<String, CardTransactionEvent> kafka;
5

6
    CardTransactionPublisher(KafkaTemplate<String, CardTransactionEvent> kafka) {
7
        this.kafka = kafka;
8
    }
9

10
    void publish(CardTransactionEvent event) {
11
        // Key = account: a purchase and its later reversal land on the same partition,
12
        // so they are consumed in the order they happened.
13
        kafka.send("card-transactions", event.accountId(), event);
14
    }
15
}

1
@Component
2
class BalanceProjectionListener {
3

4
    private final ProcessedEvents processed;
5
    private final AvailableBalances balances;
6

7
    BalanceProjectionListener(ProcessedEvents processed, AvailableBalances balances) {
8
        this.processed = processed;
9
        this.balances = balances;
10
    }
11

12
    // The offset is committed after this method returns. If the pod dies before that,
13
    // the event comes again, and the processed_event row makes the second run a no-op.
14
    @KafkaListener(topics = "card-transactions")
15
    @Transactional
16
    void on(CardTransactionEvent event) {
17
        if (!processed.markIfNew(event.eventId())) {
18
            return;   // already applied: a redelivery after a crash or a rebalance
19
        }
20
        switch (event.type()) {
21
            case PURCHASE -> balances.hold(event.accountId(), event.amount());
22
            case REVERSAL -> balances.release(event.accountId(), event.amount());
23
        }
24
    }
25
}

1
CREATE TABLE processed_event (
2
    event_id      UUID         PRIMARY KEY,   -- a redelivered event cannot be inserted twice
3
    processed_at  TIMESTAMPTZ  NOT NULL DEFAULT now()
4
);
5

6
-- markIfNew runs:
7
--   INSERT INTO processed_event (event_id) VALUES (?) ON CONFLICT DO NOTHING
8
-- and treats "1 row inserted" as new, "0 rows" as already processed.

1
// Counter-example: keyed by the transaction, not the account.
2
void publish(CardTransactionEvent event) {
3
    // A purchase and its reversal have different transaction ids, so they hash to
4
    // different partitions. The reversal can be consumed first and release a hold
5
    // that does not exist yet; the purchase then holds money that was already refunded.
6
    kafka.send("card-transactions", event.transactionId(), event);
7
}

Tuzaklar: kaç partition seçmeli?

Fazla partition da bedava değil: her partition broker’da dosya tanıtıcısı ve bellek tüketir, lider seçimi süresini uzatır ve uçtan uca gecikmeyi artırır.

Pratik yaklaşım: hedef throughput’u tek bir partition’ın ölçülen throughput’una böl ve büyüme payı ekle. Azaltamayacağın için biraz cömert davran, ama “ne olur ne olmaz” diye yüzlerce partition açmak gerçek bir maliyettir.

Kendini sına