Message Ordering ve Partitioning: Sıra Garantisi Nerede Başlar, Nerede Biter?

Sürüm notu: Bu yazıdaki kodlar C# 14 ve .NET 10 ile, Apache Kafka 4.2 (Docker container'ı) üzerinde derlenip denenmiştir.

Önce Tahmin Et

Ada bir sipariş veriyor, birkaç saniye sonra teslimat adresini İzmir'den Ankara'ya değiştiriyor, bir dakika sonra da siparişi iptal ediyor. Ordering üç event yayınlıyor; her biri siparişin o anki version'ını taşıyor:

using System.Text.Json.Serialization;

namespace Bookstore.Ordering.Contracts;

// Version: aynı siparişin event'leri için artan sıra numarası (aggregate'in version'ı)
[JsonPolymorphic(TypeDiscriminatorPropertyName = "type")]
[JsonDerivedType(typeof(OrderPlaced), "order-placed")]
[JsonDerivedType(typeof(ShippingAddressChanged), "shipping-address-changed")]
[JsonDerivedType(typeof(OrderCancelled), "order-cancelled")]
public abstract record OrderEvent(Guid OrderId, int Version);

public sealed record OrderPlaced(Guid OrderId, int Version, string Customer, string City) : OrderEvent(OrderId, Version);
public sealed record ShippingAddressChanged(Guid OrderId, int Version, string City) : OrderEvent(OrderId, Version);
public sealed record OrderCancelled(Guid OrderId, int Version) : OrderEvent(OrderId, Version);

Shipping bu event'lerden bir kargo planı çıkarıyor. Kampanya yüzünden Shipping'in consumer'ı iki instance'a çıkarılmış ve aynı queue'dan okuyorlar. ShippingAddressChanged'i alan instance biraz yavaş kalıyor, OrderCancelled'i alan ise hızlı. Event'ler Shipping'de 1, 3, 2 sırasıyla işleniyor. Kargo planının son hâli ne olur? Ada'nın iptal ettiği sipariş kargoya verilir mi?

Sıra Nerede Bozuluyor?

Önce iki consumer yazalım. Biri event'leri geldiği sırayla uyguluyor, diğeri version'a bakıyor:

using Bookstore.Ordering.Contracts;

namespace Bookstore.Shipping;

public enum PlanStatus { ReadyToShip, Cancelled }

public sealed class ShipmentPlan(Guid orderId)
{
    public Guid OrderId { get; } = orderId;
    public PlanStatus Status { get; set; }
    public string City { get; set; } = "";
    public int LastVersion { get; set; }
    public override string ToString() => $"{Status}, {City}, v{LastVersion}";
}

// Sıraya güvenen consumer: event hangi sırayla gelirse o sırayla uygular
public sealed class NaiveShipmentPlanner
{
    public Dictionary<Guid, ShipmentPlan> Plans { get; } = [];

    public void Apply(OrderEvent e)
    {
        ShipmentPlan plan = Plans.TryGetValue(e.OrderId, out ShipmentPlan? p) ? p : Plans[e.OrderId] = new(e.OrderId);
        switch (e)
        {
            case OrderPlaced placed: plan.City = placed.City; plan.Status = PlanStatus.ReadyToShip; break;
            case ShippingAddressChanged moved: plan.City = moved.City; plan.Status = PlanStatus.ReadyToShip; break;
            case OrderCancelled: plan.Status = PlanStatus.Cancelled; break;
        }
        plan.LastVersion = e.Version;
    }
}

// Sıraya güvenmeyen consumer: version'ı eski olan event'i reddeder
public sealed class VersionedShipmentPlanner
{
    public Dictionary<Guid, ShipmentPlan> Plans { get; } = [];
    public List<string> Rejected { get; } = [];

    public void Apply(OrderEvent e)
    {
        ShipmentPlan plan = Plans.TryGetValue(e.OrderId, out ShipmentPlan? p) ? p : Plans[e.OrderId] = new(e.OrderId);
        if (e.Version <= plan.LastVersion)
        {
            Rejected.Add($"{e.GetType().Name} v{e.Version} (already at v{plan.LastVersion})");
            return;
        }
        switch (e)
        {
            case OrderPlaced placed: plan.City = placed.City; break;
            case ShippingAddressChanged moved: plan.City = moved.City; break;
            case OrderCancelled: plan.Status = PlanStatus.Cancelled; break;
        }
        plan.LastVersion = e.Version;
    }
}

using System.Text.Json;
using Bookstore.Ordering.Contracts;
using Bookstore.Shipping;
using Confluent.Kafka;
using Confluent.Kafka.Admin;

// Aynı siparişin üç event'i, iki competing consumer yüzünden karışık sırayla işleniyor
Guid adaOrder = Guid.CreateVersion7();
OrderEvent[] published =
[
    new OrderPlaced(adaOrder, 1, "Ada", "Izmir"),
    new ShippingAddressChanged(adaOrder, 2, "Ankara"),
    new OrderCancelled(adaOrder, 3),
];
OrderEvent[] processed = [published[0], published[2], published[1]];   // v2, v3'ten sonra işlendi

NaiveShipmentPlanner naive = new();
VersionedShipmentPlanner versioned = new();
foreach (OrderEvent e in processed) { naive.Apply(e); versioned.Apply(e); }
Console.WriteLine($"[naive]     {naive.Plans[adaOrder]}");
Console.WriteLine($"[versioned] {versioned.Plans[adaOrder]}; rejected: {string.Join(", ", versioned.Rejected)}");
Console.WriteLine();

[naive]     ReadyToShip, Ankara, v2
[versioned] Cancelled, Izmir, v3; rejected: ShippingAddressChanged v2 (already at v3)

Sıraya güvenen consumer'ın son hâli "kargoya hazır, Ankara". İptal edilmiş bir sipariş kargoya verilecek. Çünkü adres değişikliği event'inin handler'ı, iptalden sonra gelince planı yeniden "hazır"a çekti. Version'a bakan consumer ise v2'yi "zaten v3'teyim" diyerek reddetti ve sipariş iptal olarak kaldı.

Bu karışıklık için bir bug gerekmedi; önceki iki yazıdaki iki doğal davranış yetti. Competing consumer'lar message'ları paralel işler ve hangisinin önce bitireceği belli değildir. At-least-once teslimde de bir message yeniden teslim edildiğinde arkasındaki message'lar ondan önce işlenmiş olabilir. Microsoft'un Queue-Based Load Leveling rehberi de aynı uyarıyı yapıyor: birden çok consumer paralel çalıştığında queue, gönderim sırasını her koşulda korumaz. Enterprise Integration Patterns kitabı (EIP: Gregor Hohpe ve Bobby Woolf, 2003) messaging'in zorlukları arasında sıra sorunlarını sayarken tam bunu kastediyordu.

Global Sıra Yok, Anahtar Başına Sıra Var

Sistemde bütün message'ların tek bir global sırada işlenmesini istiyorsan, tek bir consumer'la tek bir sıradan okuman gerekir: paralellik yok, ölçek yok. Neyse ki çoğu zaman global sıraya ihtiyaç yok. Ada'nın siparişinin event'leri birbirine göre sıralı olmalı; Ada'nın siparişiyle Alan'ın siparişinin event'lerinin göreli sırası kimseyi ilgilendirmiyor. İhtiyaç, anahtar başına sıra: aynı aggregate'in event'leri sırayla, farklı aggregate'ler paralel.

Broker'lar bunu farklı adlarla sunuyor. Kafka'da partition key: aynı key'e sahip kayıtlar aynı partition'a gider ve bir partition kendi içinde sıralıdır; consumer group'ta bir partition'ı aynı anda tek bir consumer okur. Azure Service Bus'ta session: aynı SessionId'ye sahip message'lar FIFO (first in, first out) sırasıyla ve aynı anda yalnızca o session'ın kilidini tutan tek bir receiver'a teslim edilir. RabbitMQ'da single active consumer: bir queue'ya birden çok consumer bağlı olsa da yalnızca biri aktiftir, diğerleri yedekte bekler. Bu sırayı korur ama paralelliği de queue başına bire indirir; paralellik için RabbitMQ'nun consistent hash exchange'i ya da super stream'leri ile anahtar başına ayrı queue/stream kurulur.

Kafka'da Partition Key

Bookstore'un Ordering event'lerini tek bir topic'e, siparişin id'si key olacak şekilde yazalım. Karşılaştırma için bir siparişin event'lerini de yanlış bir key'le, her event'in kendi id'siyle yazalım:

// ---------- Kafka: key = OrderId, aynı siparişin event'leri aynı partition'da ----------
const string bootstrap = "localhost:9092";
const string topic = "eda05.orders";
using IAdminClient admin = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrap }).Build();
await admin.CreateTopicsAsync([new TopicSpecification { Name = topic, NumPartitions = 3, ReplicationFactor = 1 }]);

ProducerConfig producerConfig = new()
{
    BootstrapServers = bootstrap,
    Partitioner = Partitioner.Murmur2Random,   // Java producer'ıyla aynı hash
};
using IProducer<string, string> producer = new ProducerBuilder<string, string>(producerConfig).Build();

async Task<int> PublishAsync(string key, OrderEvent e, IProducer<string, string>? via = null)
{
    DeliveryResult<string, string> result = await (via ?? producer).ProduceAsync(topic,
        new Message<string, string> { Key = key, Value = JsonSerializer.Serialize(e) });
    return result.Partition.Value;
}

string[] customers = ["Ada", "Alan", "Grace"];
Dictionary<string, Guid> orderOf = customers.ToDictionary(c => c, _ => Guid.CreateVersion7());
foreach (string c in customers)
{
    Guid id = orderOf[c];
    OrderEvent[] events = [new OrderPlaced(id, 1, c, "Izmir"), new ShippingAddressChanged(id, 2, "Ankara"), new OrderCancelled(id, 3)];
    List<int> partitions = [];
    foreach (OrderEvent e in events) partitions.Add(await PublishAsync(key: id.ToString(), e));
    Console.WriteLine($"key=OrderId    {c,-6} v1..v3 -> partitions {string.Join(",", partitions)}");
}

// Yanlış key: her event'in kendi id'si
{
    Guid id = Guid.CreateVersion7();
    OrderEvent[] events = [new OrderPlaced(id, 1, "Linus", "Izmir"), new ShippingAddressChanged(id, 2, "Ankara"), new OrderCancelled(id, 3)];
    List<int> partitions = [];
    foreach (OrderEvent e in events) partitions.Add(await PublishAsync(key: Guid.NewGuid().ToString(), e));
    Console.WriteLine($"key=MessageId  Linus  v1..v3 -> partitions {string.Join(",", partitions)}");
}
Console.WriteLine();

key=OrderId    Ada    v1..v3 -> partitions 2,2,2
key=OrderId    Alan   v1..v3 -> partitions 2,2,2
key=OrderId    Grace  v1..v3 -> partitions 2,2,2
key=MessageId  Linus  v1..v3 -> partitions 2,0,2

Key OrderId olduğunda her siparişin üç event'i aynı partition'a gitti; bu koşuda üç sipariş tesadüfen aynı partition'a düştü, ama her biri kendi içinde sıralı. Key MessageId olduğunda Linus'un siparişinin v2'si başka bir partition'a gitti. Artık onu başka bir consumer işleyebilir ve v1'den önce ya da v3'ten sonra işlenmesine hiçbir şey engel değil. Partition'lar, consumer'lar ve key seçimi arasındaki ilişki:

Koddaki Partitioner.Murmur2Random satırı bir tuzaktan korunmak için orada. Confluent.Kafka'nın altındaki librdkafka'nın varsayılan partitioner'ı key'in CRC32 (Cyclic Redundancy Check) hash'ini kullanıyor; Java producer'ının varsayılanı ise murmur2. Aynı topic'e hem .NET hem Java servisleri yazıyorsa ve ayar yapılmadıysa, aynı sipariş id'si iki dilde farklı partition'lara gidebilir ve anahtar başına sıra sessizce bozulur. librdkafka dokümanı murmur2_random'ı Java'nın varsayılanının işlevsel karşılığı olarak tanımlıyor.

Producer tarafında da sıranın bir koşulu var: retry. Bir kayıt geçici bir hatayla başarısız olup tekrar gönderilirken arkasındaki kayıt yazılmışsa sıra değişebilir. Kafka'nın idempotent producer'ı (bir önceki yazıda gördüğümüz, sıra numaralı gönderim) bu durumu da kapatıyor: idempotence açıkken bağlantı başına beşe kadar eşzamanlı istek sırayı bozmadan gönderilebiliyor. .NET tarafında bunun için EnableIdempotence = true ayarı gerekiyor; librdkafka'da varsayılanı kapalı.

Key Seçimi ve Hot Partition

Key'in iki görevi var: sırayı belirlemek ve yükü dağıtmak. Kampanya döneminde siparişlerin büyük kısmı tek bir kitaba gidiyorsa ve biri "stokla ilgili event'leri kitaba göre sıralayalım" deyip key'i BookId yaptıysa ne olur?

// ---------- Hot partition: key seçimi yükün dağılımını belirler ----------
string[] books = ["Dune", "Emma", "The Hobbit", "Middlemarch"];
int[] byBook = new int[3], byOrder = new int[3];
for (int i = 0; i < 1000; i++)
{
    string book = i % 10 < 7 ? "Dune" : books[i % 4];   // kampanyada siparişlerin ~%70'i Dune
    Guid id = Guid.CreateVersion7();
    OrderPlaced e = new(id, 1, "Ada", "Izmir");
    byBook[await PublishAsync(key: book, e)]++;
    byOrder[await PublishAsync(key: id.ToString(), e)]++;
}
Console.WriteLine($"key=BookId   per partition: {string.Join(" / ", byBook)}");
Console.WriteLine($"key=OrderId  per partition: {string.Join(" / ", byOrder)}");
Console.WriteLine();

key=BookId   per partition: 850 / 100 / 50
key=OrderId  per partition: 335 / 343 / 322

Aynı bin event, key BookId iken partition'lara 850 / 100 / 50 dağıldı: Dune'un bütün event'leri tek partition'a gitti. O partition'ı okuyan tek consumer bütün kampanya yükünü taşırken diğerleri neredeyse boşta. Consumer eklemek de çözmez, çünkü bir partition'ı aynı anda tek consumer okuyabilir. Key OrderId iken dağılım 335 / 343 / 322; siparişler birbirinden bağımsız olduğu için yük dengeli. Buna hot partition deniyor ve kaynağı hemen her zaman, dağılımı eşit olmayan bir key.

Bookstore için karar: Ordering event'lerinde key OrderId. Sıra ihtiyacı sipariş aggregate'inin sınırıyla örtüşüyor ve siparişler eşit dağılıyor. CustomerId çekici görünebilir ("müşterinin siparişleri sıralı olsun"), ama 400 kitaplık sipariş veren bir okul, kendi partition'ını ısıtır. Domain-Driven Design serisinin aggregate yazılarındaki kural burada da işe yarıyor: tutarlılık sınırı aggregate'tir, sıra ihtiyacı da çoğu zaman aggregate'le biter.

Partition Sayısını Değiştirmek

Partition sayısı paralelliğin üst sınırıysa, yük artınca partition eklemek akla gelen ilk çözüm. Ama key'den partition'a giden yol bir hash ve bölümü: hash(key) % partitionCount. Bölen değişirse sonuç da değişir:

// ---------- Partition sayısını artırmak: aynı key başka partition'a gidebilir ----------
Guid[] openOrders = [.. Enumerable.Range(0, 12).Select(_ => Guid.CreateVersion7())];
Dictionary<Guid, int> before = [];
foreach (Guid id in openOrders) before[id] = await PublishAsync(id.ToString(), new OrderPlaced(id, 1, "Ada", "Izmir"));
await admin.CreatePartitionsAsync([new PartitionsSpecification { Topic = topic, IncreaseTo = 4 }]);
// Çalışan producer metadata'yı periyodik yeniler (varsayılan 5 dk); yeni deploy edilmiş bir instance'ı taklit edelim
using IProducer<string, string> redeployed = new ProducerBuilder<string, string>(producerConfig).Build();
List<string> moved = [];
foreach (Guid id in openOrders)
{
    int after = await PublishAsync(id.ToString(), new OrderCancelled(id, 2), redeployed);
    if (after != before[id]) moved.Add($"p{before[id]}->p{after}");
}
Console.WriteLine($"3 -> 4 partitions: {moved.Count} of {openOrders.Length} orders moved ({string.Join(", ", moved)})");

await admin.DeleteTopicsAsync([topic]);

3 -> 4 partitions: 11 of 12 orders moved (p0->p2, p1->p2, p0->p1, p2->p1, p2->p0, p2->p3, p2->p1, p2->p0, p0->p2, p1->p2, p2->p0)

On iki açık siparişin on birinin sonraki event'leri başka bir partition'a gitti. Bir siparişin eski event'leri hâlâ eski partition'da işlenmeyi beklerken yenileri yeni partition'da, başka bir consumer tarafından işlenebilir; geçiş anında sipariş başına sıra garantisi yoktur. İlk denememde partition sayısını artırıp üç saniye bekledim ve aynı producer'la gönderdim; hiçbir key taşınmadı, çünkü çalışan producer topic'in yeni partition sayısını henüz öğrenmemişti (librdkafka metadata'yı varsayılan olarak beş dakikada bir yeniliyor). Yani geçiş anında farklı producer instance'ları aynı key'i farklı partition'lara bile yazabilir. Pratik sonuç: partition sayısını baştan, beklenen yükün biraz üstünde seç ve sonradan artırmayı sıra açısından bir migration olarak ele al.

Sıraya Bağımlı Olmamak

Partition key ve session'lar sırayı broker'da korur, ama consumer'ın kendi içindeki paralellik, retry'lar ve dead-letter'dan yeniden işleme yine de sırayı bozabilir. Daha sağlam yol, consumer'ı sıraya ihtiyaç duymayacak şekilde tasarlamak. Üç teknik var.

Version ya da sequence alanı. Yukarıdaki VersionedShipmentPlanner bunu yapıyor: her event, aggregate'in version'ını taşıyor ve consumer bildiğinden eski bir version'ı reddediyor. Domain-Driven Design serisinin, aggregate'leri Entity Framework (EF) Core ile saklamayı anlatan "Aggregate'leri EF Core 10 ile Persist Etmek" yazısında Order.Version'ı concurrency token olarak eklemiştik ve her değişiklikte artıyordu; aynı sayı event'e konunca consumer için hazır bir sıra numarası oluyor. Bu yaklaşım event'in ilgili state'i taşıdığı durumda (serinin 02. yazısındaki event-carried state transfer) en iyi çalışır: v3 geldiğinde v2'nin içeriğine ihtiyacın yoktur. Event'ler yalnızca farkı taşıyorsa, aradaki boşluğu fark edip eksik event'i bekletmek ya da state'i kaynaktan okumak gerekir.

Commutative update. Sırası değiştiğinde sonucu değişmeyen işlemler sıraya ihtiyaç duymaz. Stok için "+2", "-1" gibi farklar hangi sırayla uygulanırsa uygulansın aynı toplamı verir. Durum geçişleri ise commutative değildir: "iptal edildi" ile "adres değişti"nin sırası sonucu değiştirir. Bir dikkat: commutative bir fark, idempotent değildir; "+2" iki kez uygulanırsa stok bozulur. Serinin bir sonraki yazısının konusu tam bu.

State machine'i consumer'a koymak. Shipping, "iptal edilmiş bir plan yeniden hazıra dönemez" kuralını kendi tarafında uygularsa, event'ler hangi sırayla gelirse gelsin iptal kazanır. Bu, sırayı değil, domain kuralını koruyan bir savunma.

Trade-off: Sıra, Paralellik, Throughput

Bookstore'un seçimi ikinci satır artı üçüncü satırın disiplini: Ordering event'leri OrderId key'iyle partition'lanıyor, Shipping ve Inventory de event'lerdeki version'ı kontrol ediyor. Broker'ın sırası bir optimizasyon, consumer'ın kontrolü ise güvenlik ağı.

Sık Düşülen Tuzaklar

Key vermemek ya da message id'yi key yapmak

Key'siz kayıtlar partition'lara dağıtılır ve aynı siparişin event'leri farklı partition'lara gidebilir. Her event'in kendi id'sini key yapmak da aynı sonucu verir; Linus'un siparişinde gördüğümüz bu. Key, sıranın korunmasını istediğin şeyin kimliği olmalı.

Aynı aggregate'in event'lerini farklı topic'lere yazmak

OrderPlaced bir topic'e, OrderCancelled başka bir topic'e yazılırsa, ikisi arasında hiçbir sıra garantisi kalmaz; aynı key'i kullansalar bile. Sırası birbirine bağlı event'ler aynı topic'te ve aynı key'le olmalı.

Consumer içinde paralelleştirmek

Partition'dan sırayla okuyup her message'ı Task.Run ile ayrı bir task'e vermek, broker'ın koruduğu sırayı consumer'ın içinde bozar. Paralellik gerekiyorsa key başına sıralı işleyen bir yapı kur (key'e göre bölünmüş birden çok Channel<T> gibi) ya da partition sayısını artır.

Sıra garantisini consumer'ın tek savunması saymak

Session ya da partition key, dead-letter queue'dan saatler sonra yeniden gönderilen bir message'ı korumaz: o message, kendisinden sonraki bütün event'lerden sonra gelir. Version kontrolü olmayan consumer, eski bir state'i yeniden uygular.

Küçük Bir Deneme

★☆☆ processed dizisini [published[2], published[0], published[1]] yap: iptal, siparişten önce geliyor. İki planner'ın sonuçları ne oluyor? Version'a bakan planner için bu sonuç doğru mu?

★★☆ NaiveShipmentPlanner'a version eklemeden yalnızca bir domain kuralı ekle: iptal edilmiş bir plan başka bir event'le yeniden ReadyToShip olamaz. Açılıştaki senaryoda sonuç doğru çıkıyor mu? Hangi senaryoda hâlâ yanlış kalır?

★★★ Producer'da Partitioner satırını kaldır (librdkafka varsayılanı) ve aynı on iki sipariş id'sinin hangi partition'lara gittiğini Murmur2Random ile karşılaştır. Kaç tanesi farklı partition'a düşüyor? Bunu bir Java servisinin aynı topic'e yazdığı bir sistemde nasıl fark ederdin?

Version kontrolü eski bir event'i reddetti; ama aynı event iki kez gelirse? v3 ikinci kez geldiğinde reddedilir, çünkü version'ı eski değil, eşit. Peki event bir stok farkı (+2) ya da bir ödeme command'ı olsaydı? Bir sonraki yazıda idempotent consumer'a bakacağız: aynı message iki kez geldiğinde ne olur ve dedup kaydının iş verisiyle aynı transaction'da olması neden şart?