Confluent.Kafka ile .NET'te Kafka: Producer, Consumer Group, Offset Commit ve Rebalance

Sürüm notu: Bu yazıdaki kodlar .NET 10, Apache Kafka 4.2.1 (KRaft) ve Confluent.Kafka 2.16.0 ile derlenip denenmiştir.

Önce Tahmin Et

Bir Kafka consumer'ı varsayılan ayarlarla çalışıyor; tek fark auto commit aralığının 1 saniyeye indirilmiş olması. Her mesajı işlemek 700 ms sürüyor. Consumer partition 1'deki 3 numaralı offset'i işlerken process çöküyor. Consumer group yeniden başladığında partition 1'i hangi offset'ten okumaya devam eder: 3'ten mi, 4'ten mi? Yani yarıda kalan fatura tekrar mı kesilir, yoksa hiç mi kesilmez?

İkinci soru: Produce ile 100 mesaj gönderip using bloğunun sonunda producer'ı dispose ediyorsun. Kaç mesaj broker'a ulaşır?

Kafka'da Queue Yok, Log Var

RabbitMQ ve Azure Service Bus'ta consumer mesajı alır, işler ve onaylar; onaylanan mesaj silinir. Kafka'da topic, partition'lara bölünmüş, yalnızca sona eklenen bir log'dur. Mesaj okununca silinmez; retention süresi dolana kadar durur. Consumer'ın yönettiği şey mesaj değil, her partition'daki okuma konumu yani offset. Aynı topic'i okuyan her consumer group kendi offset'lerini tutar; Billing ve Notification aynı mesajları birbirinden habersiz okur. Bir group içindeki consumer'lar ise partition'ları paylaşır: bir partition aynı anda group'taki tek bir consumer'a atanır.

Bunun .NET kodundaki üç sonucu var. Birincisi, mesaj başına ack yok; "buraya kadar okudum" diye offset commit edersin. İkincisi, retry ve dead-letter Kafka'nın özelliği değil, senin kodunun işi. Üçüncüsü, sıralama yalnızca partition içinde garanti; paralellik de partition sayısıyla sınırlı. Sıralama ve partitioning'in kavramsal tarafı Messaging ve Event-Driven Architecture serisinin "Message Ordering ve Partitioning" yazısında. Örnek ortam: bookstore.orders topic'i, 3 partition, tek broker'lı Kafka 4.2.1 container'ı (ZooKeeper'sız KRaft modu).

Producer: Key, Acks ve Idempotence

public sealed class OrderEventProducer : IDisposable
{
    private readonly IProducer<string, string> _producer;

    public OrderEventProducer(string bootstrapServers)
    {
        ProducerConfig config = new()
        {
            BootstrapServers = bootstrapServers,
            ClientId = "bookstore-order-api",
            Acks = Acks.All,
            EnableIdempotence = true,
            LingerMs = 5,
            // Java client'larıyla aynı partition'ı seçmek için
            Partitioner = Partitioner.Murmur2Random
        };
        _producer = new ProducerBuilder<string, string>(config).Build();
    }

    public async Task<DeliveryResult<string, string>> PublishAsync(OrderPlaced order, CancellationToken ct)
    {
        Message<string, string> message = new()
        {
            // aynı key her zaman aynı partition'a gider: sipariş bazında sıra
            Key = order.OrderId.ToString(),
            Value = JsonSerializer.Serialize(order, MessageJson.Options),
            Headers = new Headers
            {
                { "message-id", Encoding.UTF8.GetBytes(Guid.NewGuid().ToString()) },
                { "type", Encoding.UTF8.GetBytes(OrderPlaced.MessageType) }
            }
        };
        return await _producer.ProduceAsync("bookstore.orders", message, ct);
    }

    public void Dispose()
    {
        // tampondaki mesajlar gönderilmeden kapatma
        _producer.Flush(TimeSpan.FromSeconds(10));
        _producer.Dispose();
    }
}

IProducer thread-safe ve pahalı bir object: arkasında librdkafka'nın kendi thread'leri, bağlantıları ve gönderim tamponu var. Uygulama ömrü boyunca tek instance (singleton) olarak yaşamalı. Altı sipariş ve aynı siparişin iki tekrarı için delivery report'lar:

key=c17ccad4 -> partition 0, offset 2, status Persisted
key=b183dc39 -> partition 2, offset 0, status Persisted
key=ad6a3f91 -> partition 1, offset 6, status Persisted
key=70097c09 -> partition 0, offset 3, status Persisted
key=061b6a13 -> partition 0, offset 4, status Persisted
key=e60d9c0c -> partition 1, offset 7, status Persisted
key=c17ccad4 -> partition 0, offset 5 (same key again)
key=c17ccad4 -> partition 0, offset 6 (same key again)

Partition'ı key'in hash'i belirliyor; aynı key tekrar aynı partition'a, artan offset'lerle gitti. Bir siparişin bütün event'lerini sipariş id'si ile key'lersen o siparişin event'leri sırayla okunur. Key'siz mesajlar partition'lara dağıtılır ve aralarında sıra kalmaz.

Ayarların gerekçesi:

• Acks = Acks.All: broker, mesajı in-sync replikaların hepsine yazınca onaylar. librdkafka'da varsayılan zaten bu, ama açıkça yazmak niyeti belgeler.
• EnableIdempotence = true: producer'ın kendi retry'ları partition'da kopya ya da sıra bozulması üretmez. librdkafka'da varsayılanı false. Bu, producer tarafındaki tekrarları çözer; consumer tarafındaki tekrarları çözmez.
• LingerMs: mesajı göndermeden önce batch dolsun diye beklenen süre; librdkafka varsayılanı 5 ms. Yüksek hacimde artırmak throughput'u yükseltir, gecikmeyi artırır.
• Partitioner = Murmur2Random: librdkafka'nın varsayılan partitioner'ı consistent_random ve key'i CRC32 (Cyclic Redundancy Check) ile hash'liyor; Java client'ı ise murmur2 kullanıyor. Aynı topic'e hem .NET hem Java servisleri yazıyorsa aynı key iki farklı partition'a gidebilir ve sipariş bazındaki sıra sessizce bozulur. Murmur2Random, librdkafka dokümantasyonuna göre Java producer'ının varsayılan partitioner'ıyla işlevsel olarak eşdeğer.

Flush ve Dispose

ProduceAsync delivery report gelene kadar bekler; Produce ise mesajı tampona koyup hemen döner, sonucu callback ile bildirir. Yüksek hacimde Produce çok daha hızlıdır, ama tampondaki mesajlar henüz gönderilmemiştir. Yazının başındaki ikinci soru, LingerMs = 1000 ile:

%4|1794104763.788|TERMINATE|rdkafka#producer-1| [thrd:app]: Producer terminating with 100 messages (100 bytes) still in queue or transit: use flush() to wait for outstanding message delivery
flush=False: delivered 0 of 100
Flush returned, still in queue: 0
flush=True: delivered 100 of 100

Flush çağrılmadan dispose edilen producer 100 mesajın hiçbirini göndermedi. librdkafka uyarıyı stderr'e yazdı ama exception yok. Flush(timeout) tamponda kalan mesaj sayısını döner; sıfır değilse loglamalısın. Graceful shutdown'da producer'ı dispose etmeden önce flush etmek bu yüzden zorunlu.

Consumer'ı BackgroundService İçinde Çalıştırmak

public sealed class OrderPlacedConsumer(
    IConfiguration configuration, InvoiceService invoices, ILogger<OrderPlacedConsumer> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        ConsumerConfig config = new()
        {
            BootstrapServers = configuration["Kafka:BootstrapServers"],
            GroupId = "billing",
            ClientId = $"billing-{Environment.ProcessId}",
            AutoOffsetReset = AutoOffsetReset.Earliest,
            EnableAutoCommit = true,
            // offset'i işlem bittikten sonra biz işaretleyeceğiz
            EnableAutoOffsetStore = false
        };

        using IConsumer<string, string> consumer = new ConsumerBuilder<string, string>(config)
            .SetPartitionsAssignedHandler((_, partitions) =>
                logger.LogInformation("Assigned: {Partitions}", string.Join(", ", partitions.Select(p => p.Partition.Value))))
            .SetPartitionsRevokedHandler((_, partitions) =>
                logger.LogInformation("Revoked: {Partitions}", string.Join(", ", partitions.Select(p => p.Partition.Value))))
            .SetErrorHandler((_, error) => logger.LogWarning("Kafka error: {Reason}", error.Reason))
            .Build();

        consumer.Subscribe("bookstore.orders");

        try
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                // Consume senkron ve bloklayıcı: mesaj gelene ya da token iptal edilene kadar bekler
                ConsumeResult<string, string> result = consumer.Consume(stoppingToken);

                OrderPlaced order = JsonSerializer.Deserialize<OrderPlaced>(result.Message.Value, MessageJson.Options)!;
                await invoices.CreateInvoiceAsync(order, stoppingToken);

                // yalnızca işlenmiş mesajın offset'i commit'e aday olur
                consumer.StoreOffset(result);
                logger.LogInformation("Stored offset {Partition}@{Offset}",
                    result.Partition.Value, result.Offset.Value);
            }
        }
        catch (OperationCanceledException)
        {
            // normal kapanış
        }
        finally
        {
            // saklanan offset'leri commit eder ve group'tan düzgünce ayrılır
            consumer.Close();
            logger.LogInformation("Consumer closed");
        }
    }
}

RabbitMQ ve Service Bus client'ları mesajı push eden bir event ile çalışıyordu. Confluent.Kafka'nın consumer'ı pull modelinde ve Consume senkron, bloklayıcı bir çağrı; async sürümü yok. Bu döngü bir thread'i sürekli meşgul eder. .NET 10'da BackgroundService.ExecuteAsync'in tamamı arka plan thread'inde çalıştığı için bu döngü host'un açılışını engellemiyor. .NET 8 ve 9'da ise ExecuteAsync'in ilk await'e kadarki kısmı startup'ta senkron çalışıyordu; ilk satırı bloklayan bir Consume döngüsü, diğer hosted service'lerin ve Kestrel'in hiç başlamamasına yol açabiliyordu. Eski sürümlerde döngüyü Task.Run ile ayrı başlatmak bu yüzden yaygın bir alışkanlık. Consumer instance'ı thread-safe değil; aynı anda tek thread'den çağır.

consumer.Close() önemli: saklanmış offset'leri commit eder ve group coordinator'a "ayrılıyorum" der. Close edilmeden ölen consumer'ın partition'ları, broker onun öldüğünü session timeout ile anlayana kadar (librdkafka varsayılanı 45 saniye) kimseye atanmaz.

Offset Commit: Auto Commit Tuzağı ve At-Least-Once

librdkafka'nın varsayılanları: enable.auto.commit = true, auto.commit.interval.ms = 5000, enable.auto.offset.store = true. Son ayar, Consume'un döndürdüğü her mesajın offset'inin, işlenip işlenmediğine bakılmadan commit'e aday olması demek. İlk sorunun cevabı:

0.2s processing 2@0
0.9s processing 1@0
1.6s processing 1@1
2.3s processing 1@2
3.0s processing 1@3
3.7s crash while processing 1@3
exit=1

GROUP           TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
lab-autocommit  bookstore.orders 0          -               7               -
lab-autocommit  bookstore.orders 1          4               8               4
lab-autocommit  bookstore.orders 2          1               1               0

Kafka'da commit edilen offset "bir sonraki okunacak mesaj"dır. Partition 1 için commit edilen değer 4: group yeniden başladığında 3 numaralı mesajı hiç görmeyecek. İşlenmekte olan sipariş, hiçbir hata kaydı olmadan atlandı. Bu at-most-once davranışı; Messaging ve Event-Driven Architecture serisinin "Delivery Guarantee'leri" yazısındaki kavramın Kafka'daki gerçek hâli.

Çözüm Billing kodunda: EnableAutoOffsetStore = false ve işlem bittikten sonra StoreOffset(result). Auto commit açık kalır ama yalnızca bizim işaretlediğimiz offset'leri commit eder. Bunun bedeli tekrar. Altı yeni siparişi işleyen Billing'i kill -9 ile öldürdük; işlem bitti, offset'ler saklandı ama 5 saniyelik commit aralığı dolmadı:

      Assigned: 0, 1, 2
      Stored offset 0@7
      Stored offset 0@8
      Stored offset 1@8
      Stored offset 2@1
      Stored offset 2@2
      Stored offset 2@3

TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG
bookstore.orders 0 7 9 2
bookstore.orders 1 8 9 1
bookstore.orders 2 1 4 3

Commit edilen offset'ler hiç ilerlememiş. Yeniden başlatılan Billing'in logu birebir aynı altı satırı tekrar yazdı: altı fatura iki kez kesildi. At-least-once budur: kayıp yok, tekrar var. Commit aralığını küçültmek pencereyi daraltır ama kapatmaz; her mesajdan sonra senkron Commit ise throughput'u ciddi düşürür. Doğru cevap consumer'ı idempotent yapmak; serinin idempotent consumer yazısında Billing'e bunu ekleyeceğiz.

Consumer Group ve Rebalance

Billing'in ilk instance'ı çalışırken ikinciyi başlattık, sonra ikinciyi durdurduk. İlk instance'ın logu (aradaki altı satır kısaltıldı):

      Assigned: 0, 1, 2
      Stored offset 2@0
      Stored offset 1@0
      Stored offset 1@1
      Stored offset 1@2
      Revoked: 0, 1, 2
      Assigned: 0, 1
      Stored offset 0@0
      ...
      Stored offset 1@3
      Revoked: 0, 1
      Assigned: 0, 1, 2
      Stored offset 1@4

İkinci instance Assigned: 2 aldı ve kapanırken Revoked: 2 verdi. Dikkat edilecek iki nokta var. Birincisi, partition 1'de tekrar yok: revoke sırasında 1@2'ye kadar saklanan offset'ler commit edildi ve yeniden atamadan sonra okuma 1@3'ten devam etti. İkincisi, ikinci instance katıldığında ilk instance yalnızca partition 2'yi değil, üçünü birden bıraktı. Bu, librdkafka'nın varsayılan range,roundrobin stratejisinin "eager" davranışı: her rebalance'ta herkes her şeyi bırakır ve yeniden dağıtım beklenir. PartitionAssignmentStrategy = CooperativeSticky yalnızca el değiştiren partition'ları bırakır.

Deney sırasında kafka-consumer-groups.sh --describe çıktısı (consumer kimliği sütunları çıkarıldı) consumer lag'i gösterir:

GROUP           TOPIC            PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
billing         bookstore.orders 0          3               7               4
billing         bookstore.orders 1          3               8               5
billing         bookstore.orders 2          1               1               0

Lag, son yazılan offset ile group'un commit ettiği offset arasındaki fark ve Kafka consumer'ının en önemli sağlık metric'i. Sürekli büyüyorsa consumer'lar yetişemiyor. Ama instance eklemenin bir tavanı var: 3 partition'lı bir topic'te dördüncü consumer hiçbir partition almaz ve boşta bekler.

Kafka 4.0 ile KIP (Kafka Improvement Proposal) 848'de tanımlanan yeni consumer protokolü genel kullanıma açıldı; Confluent.Kafka'da 2.12'den beri production-ready, ama varsayılan hâlâ classic. Bu protokolde partition ataması broker tarafında yapılır ve rebalance'lar artımlıdır. Açmak tek satır:

ConsumerConfig config = new()
{
    BootstrapServers = Bootstrap, GroupId = "billing-848", AutoOffsetReset = AutoOffsetReset.Earliest,
    GroupProtocol = GroupProtocol.Consumer
};

Assigned: 0, 1, 2
Consumed 23 messages with group.protocol=consumer
---
InvalidOperationException: `session.timeout.ms` is not supported for `group.protocol=consumer` or share consumer. It is defined broker side

GROUP                            TYPE
billing-848                      Consumer
billing                          Classic

Yeni protokolde session.timeout.ms gibi bazı ayarlar artık client'ta değil broker'da; eski config'i taşırsan consumer oluşturulurken exception alırsın. Mevcut bir group'u yeni protokole geçirmeden önce Confluent'in migration rehberini oku.

Poison Mesaj ve Dead-Letter Topic

Partition 2'ye geçersiz JSON (JavaScript Object Notation) içeren tek bir mesaj yazdık. Yukarıdaki consumer'ın logu:

info: Bookstore.Billing.Kafka.OrderPlacedConsumer[0]
      Revoked: 0, 1, 2
info: Bookstore.Billing.Kafka.OrderPlacedConsumer[0]
      Consumer closed
fail: Microsoft.Extensions.Hosting.Internal.Host[9]
      BackgroundService failed
      System.Text.Json.JsonException: Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.
crit: Microsoft.Extensions.Hosting.Internal.Host[10]
      The HostOptions.BackgroundServiceExceptionBehavior is configured to StopHost. A BackgroundService has thrown an unhandled exception, and the IHost instance is stopping. To avoid this behavior, configure this to Ignore; however the BackgroundService will not be restarted.

Host durdu ve offset ilerlemedi. Container orkestratörü process'i yeniden başlatsa bile aynı offset'ten aynı mesajı okuyacak ve aynı şekilde ölecek. RabbitMQ'da nack, Service Bus'ta MaxDeliveryCount bu döngüyü kırıyordu; Kafka'da kırıcı yok, partition'ın geri kalanı bu mesajın arkasında kilitli kalır. Process'in çıkış kodunun 0 olması da ayrı bir sürpriz: orkestratör bunu "temiz kapanış" sanabilir.

Çözüm, kalıcı hataları dead-letter topic'e yazıp offset'i ilerletmek:

catch (JsonException ex)
{
    // poison mesaj partition'ı kilitlemesin: dead-letter topic'e yaz ve devam et
    await deadLetters.ProduceAsync("bookstore.orders.billing.dlt", new Message<string, string>
    {
        Key = result.Message.Key,
        Value = result.Message.Value,
        Headers = new Headers
        {
            { "error", Encoding.UTF8.GetBytes(ex.Message) },
            { "source", Encoding.UTF8.GetBytes($"{result.Topic}/{result.Partition.Value}@{result.Offset.Value}") }
        }
    }, stoppingToken);
    consumer.StoreOffset(result);
    logger.LogWarning("Dead-lettered {Partition}@{Offset}: {Error}",
        result.Partition.Value, result.Offset.Value, ex.Message);
    continue;
}

warn: Bookstore.Billing.Kafka.OrderPlacedConsumer[0]
      Dead-lettered 2@4: Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.

kafka-console-consumer.sh ile DLT'ye (dead-letter topic) baktığımızda mesaj, hata ve kaynak konumuyla birlikte orada:

error:Expected depth to be zero at the end of the JSON payload. There is an open JSON object or array that should be closed. Path: $.orderId | LineNumber: 0 | BytePositionInLine: 12.,source:bookstore.orders/2@4	bad-1	{"orderId":

Geçici hatalar (veritabanı bir süre erişilemez) için DLT'ye hemen yazmak yerine retry topic'leri ve gecikmeli yeniden deneme gerekir; bunu serinin retry yazısında kuracağız. Burada ayrım basit: tekrar denemekle düzelmeyecek hata DLT'ye, düzelebilecek hata retry'a.

Üretimde Dikkat

Partition sayısını baştan düşün. Partition sayısı bir group'taki en yüksek paralelliği belirler ve sonradan artırılabilir, ama artırınca key'lerin partition dağılımı değişir; aynı siparişin eski ve yeni event'leri farklı partition'lara düşebilir. Azaltmak mümkün değil.

max.poll.interval.ms'i bil. İki Consume çağrısı arasında bu süre (librdkafka varsayılanı 5 dakika) geçerse consumer group'tan atılır ve partition'ları başkasına verilir. Tek mesajın işi uzun sürüyorsa ya bu süreyi artır ya da işi başka bir yere devret.

Lag'i izle ve alarm kur. Partition bazında lag, Kafka'daki en erken uyarıdır. OpenTelemetry ile metric toplamayı serinin observability yazısında yapacağız.

Retention'ı consumer'larının en uzun kesinti süresine göre seç. Consumer 8 gün durursa ve retention 7 günse, ilk gün yazılan mesajlar okunmadan silinir.

Producer'ı singleton, consumer'ı thread başına tek tut. Bir process'te birden fazla consumer gerekiyorsa her birini ayrı bir döngüde, ayrı bir instance olarak çalıştır.

Sık Düşülen Tuzaklar

Varsayılan auto offset store ile at-least-once beklemek. Gördüğümüz gibi işlenmekte olan mesaj atlanabilir. EnableAutoOffsetStore = false ve işlem sonrası StoreOffset.

Flush çağırmadan dispose. Hata yok, mesaj yok.

Consume'u async sanmak. Bloklayıcı bir çağrıyı thread pool'da sonsuz döngüde çalıştırmak bir thread'i kalıcı olarak işgal eder; .NET 8/9'da startup'ı bile durdurabilir.

.NET ve Java servislerini farklı partitioner'larla aynı topic'e yazdırmak. Sıra sessizce bozulur.

Poison mesajı exception ile bırakmak. Partition kilitlenir, host durur, yeniden başlatma aynı noktaya döner.

Kafka'yı queue gibi kullanmaya çalışmak. Mesaj bazında ack, delayed redelivery ve "queue'yu boşalt" kavramları yok; bunlara ihtiyacın varsa RabbitMQ veya Service Bus daha doğru araç.

Küçük Bir Deneme

★☆☆ Billing'i üç instance olarak çalıştır, ardından dördüncüyü başlat. kafka-consumer-groups.sh --describe çıktısında dördüncü instance'ın hiç partition almadığını gör.

★★☆ PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky ile rebalance deneyini tekrarla. İkinci instance katıldığında ilk instance hangi partition'ları bırakıyor? Logu eager davranışla karşılaştır.

★★★ Billing'e bir retry topic'i ekle: geçici hata alan mesajı bookstore.orders.billing.retry'a, deneme sayısını header'da taşıyarak yaz; ayrı bir consumer bu topic'i 30 saniye gecikmeyle işlesin ve üçüncü denemeden sonra mesajı DLT'ye göndersin.

Üç broker'ın .NET client'larını gördük ve hepsinde mesaj gövdesini JSON olarak serialize ettik. Peki OrderPlaced'a yeni bir alan eklediğimizde ya da bir alanın adını değiştirdiğimizde ne olacak? Bir sonraki yazıda mesaj contract'larını, versioning'i, geriye dönük uyumluluğu ve Kafka'da Schema Registry'yi ele alıyoruz.