Consumer'da Retry ve Redelivery: Immediate, Delayed ve Exponential Backoff

Sürüm notu: Bu yazıdaki kodlar .NET 10, RabbitMQ 4.4, RabbitMQ.Client 7.2.2, Azure.Messaging.ServiceBus 7.21 (Service Bus emulator), Confluent.Kafka 2.16 (Kafka 4.2) ve MassTransit 8.5.12 ile derlenip denenmiştir.

Önce Tahmin Et

Ödeme sağlayıcısı çöktü ve Billing consumer'ı her mesajda exception alıyor. Kod hatayı yakalıyor ve mesajı "sonra tekrar dene" diye broker'a geri bırakıyor. Queue bir quorum queue, broker RabbitMQ 4.4 ve queue'da tek bir mesaj var. Catch bloğunda şu iki satırdan biri duruyor:

await channel.BasicNackAsync(ea.DeliveryTag, multiple: false, requeue: true);
// ya da
await channel.BasicRejectAsync(ea.DeliveryTag, requeue: true);

5 saniye içinde bu tek mesaj consumer'a kaç kez gelir? İki satır aynı sonucu mu verir? 5 saniyenin sonunda mesaj nerededir?

Bu yazıda retry'ı üç broker'da ve MassTransit'te kodla kuruyoruz: hangi hatada hemen, hangisinde bekleyerek, hangisinde hiç denemeyeceğimizi. At-least-once teslimatın ne olduğu ve aynı mesajın neden tekrar gelebileceği Messaging ve Event-Driven Architecture serisinin "Delivery Guarantee'leri" yazısında anlatıldı; burada kavramı değil, kodu konuşuyoruz.

Requeue Döngüsü: Nack ile Reject Aynı Şey Değil

Deneyi iki ayrı quorum queue'da çalıştırdım. Consumer her mesajı alıp hemen geri bırakıyor; çıktıda ilk birkaç teslim, 19 ile 22 arası ve her 5000'inci teslim var. Önce BasicNackAsync(requeue: true):

   0.0s delivery #1 redelivered=False x-delivery-count=-
   0.1s delivery #2 redelivered=True x-delivery-count=-
   0.1s delivery #3 redelivered=True x-delivery-count=-
   0.1s delivery #19 redelivered=True x-delivery-count=-
   0.1s delivery #20 redelivered=True x-delivery-count=-
   0.1s delivery #21 redelivered=True x-delivery-count=-
   0.1s delivery #22 redelivered=True x-delivery-count=-
   3.1s delivery #5000 redelivered=True x-delivery-count=-
   5.0s total deliveries in 5s: 8036, messages left in queue: 0

Tek mesaj 5 saniyede 8036 kez teslim edildi. Bu bir retry değil, bir döngü: CPU (Central Processing Unit) yakıyor, log'u dolduruyor, gerçek bir handler'da ödeme sağlayıcısına saniyede yüzlerce istek atıyor ve kendiliğinden hiç bitmiyor. "messages left in queue: 0" da yanıltıcı; mesaj hâlâ orada, yalnız o an consumer'da ack bekliyor. Aynı deney BasicRejectAsync(requeue: true) ile:

   0.0s delivery #1 redelivered=False x-delivery-count=-
   0.0s delivery #2 redelivered=True x-delivery-count=1
   0.0s delivery #3 redelivered=True x-delivery-count=2
   0.1s delivery #19 redelivered=True x-delivery-count=18
   0.1s delivery #20 redelivered=True x-delivery-count=19
   0.1s delivery #21 redelivered=True x-delivery-count=20
   5.0s total deliveries in 5s: 21, messages left in queue: 0

Bu kez 21 teslimden sonra döngü durdu ve mesaj gerçekten yok oldu. Sebep quorum queue'nun delivery limit'i: RabbitMQ 4.0'dan beri varsayılan değer 20. Limit aşılınca mesaj dead-letter edilir; queue'da DLX (Dead Letter Exchange) tanımlı değilse, bizim deneyde olduğu gibi, silinir. Peki nack neden sayılmadı? Quorum queue dokümanına göre 4.3'ten beri limit yalnız "başarısız" teslimleri sayan delivery-count'a göre kontrol ediliyor. AMQP (Advanced Message Queuing Protocol) 0-9-1'de basic.reject ve consumer'ın bağlantısının kopması bu sayacı artırıyor, basic.nack artırmıyor. Yani 4.4'te BasicNackAsync(requeue: true) delivery limit'e hiç takılmayan bir sonsuz döngü.

Buradan iki ders çıkıyor. Birincisi: requeue ile "anında tekrar" neredeyse hiçbir zaman doğru retry değil, çünkü geçici bir hata bir milisaniyede düzelmez. İkincisi: her iş queue'suna DLX tanımla; yoksa delivery limit'in koruması sessiz veri kaybına dönüşür. DLQ (Dead-Letter Queue) yönetimi bir sonraki yazının konusu; bu yazıdaki örneklerin hepsinde DLX tanımlı.

Hata Transient mi, Permanent mi?

Retry'ın ilk sorusu "kaç kez" değil, "denemeye değer mi". Ödeme sağlayıcısından gelen 503, timeout ya da veritabanında deadlock geçicidir (transient): aynı mesaj birkaç saniye sonra büyük ihtimalle başarılı olur. Bozuk JSON (JavaScript Object Notation), desteklenmeyen para birimi, eksik zorunlu alan kalıcıdır (permanent): mesajı bin kez denesen bin kez aynı satırda düşer ve her denemede arkasındaki sağlam mesajları bekletir. Bu ayrımı consumer'da tek bir yerde yapıyoruz:

private static bool IsTransient(Exception ex) => ex is HttpRequestException or TimeoutException
    or TaskCanceledException { InnerException: TimeoutException };

Liste bilinçli olarak dar. Tanımadığın bir exception'ı transient saymak, kalıcı bir hatayı dakikalarca retry queue'larında dolaştırmak demek. Tersi daha ucuz: yanlışlıkla kalıcı sayılan mesaj DLQ'ya düşer, alarm çalar, sebebi düzeltip replay edersin. Kendi kodunda kalıcı hataları ayrı bir exception tipiyle (ör. PermanentMessageException) fırlatmak bu kararı okunur kılar.

Retry'ı nerede yaptığına da bak. Dış servis çağrısının içindeki birkaç yüz milisaniyelik retry (Polly ya da Microsoft.Extensions.Http.Resilience) ayrı bir katman; onu ilerideki External Service Integration ve Resilience serisinde işleyeceğiz. Burada konuştuğumuz, o katman da pes ettikten sonra mesajın kendisine ne olacağı. İki katmanın deneme sayıları çarpılır: HTTP (Hypertext Transfer Protocol) client'ta 3, consumer'da 4 deneme, tek mesaj için sağlayıcıya 12 istek demek.

In-memory retry'ın gizli maliyeti

En kolay retry, handler'ı bir döngüye almak: hata olursa 200 ms bekle, tekrar çağır. Bu sürede mesaj consumer'da "unacked" durur. RabbitMQ'da bu, prefetch penceresindeki bir slotun bekleme boyunca dolu kalması demek; prefetch 10 ve her mesaj 3 kez 2 saniye bekliyorsa, sağlayıcı çöktüğü an consumer'ın tamamı uyur. Azure Service Bus'ta mesaj PeekLock ile alınır ve lock süresi (varsayılan 1 dakika) içinde tamamlanmalıdır. ServiceBusProcessor lock'u varsayılan olarak 5 dakikaya kadar yeniler; retry beklemeleri bunu aşarsa lock düşer, mesaj başka bir consumer'a gider ve aynı mesajı iki yerde birden işlersin. Kural: in-memory retry kısa ve az sayıda olsun (toplam birkaç saniye), daha uzun bekleme broker'a devredilsin. Aşağıdaki bölümler bu devri anlatıyor.

RabbitMQ'da Delayed Retry

TTL + DLX ile bekleme queue'ları

Her RabbitMQ sürümünde çalışan klasik yol, consumer'ı olmayan "bekleme" queue'ları: mesaj buraya yazılır, TTL (Time-To-Live) dolunca DLX mekanizmasıyla asıl queue'ya geri döner. Üç seviyeli üstel bir bekleme kuruyoruz: 1, 2 ve 4 saniye (gerçek hayatta 10 saniye, 1 dakika, 10 dakika gibi). Aşağıdaki görsel, denemeden sonra çalıştıracağımız üç mesajın yolculuğunu gösteriyor:

public static class RetryTopology
{
    public const string RetryExchange = "bookstore.retry";

    public static readonly (string Queue, TimeSpan Delay)[] Levels =
    [
        ("billing.order-placed.retry.1", TimeSpan.FromSeconds(1)),
        ("billing.order-placed.retry.2", TimeSpan.FromSeconds(2)),
        ("billing.order-placed.retry.3", TimeSpan.FromSeconds(4))
    ];

    public static async Task DeclareAsync(IChannel channel)
    {
        await channel.ExchangeDeclareAsync("bookstore.ordering", ExchangeType.Topic, durable: true);
        await channel.ExchangeDeclareAsync("bookstore.dlx", ExchangeType.Direct, durable: true);
        await channel.ExchangeDeclareAsync(RetryExchange, ExchangeType.Direct, durable: true);

        // ana queue: reject(requeue:false) edilen mesaj bookstore.dlx üzerinden DLQ'ya gider
        await channel.QueueDeclareAsync("billing.order-placed", durable: true, exclusive: false, autoDelete: false,
            arguments: new Dictionary<string, object?>
            {
                ["x-queue-type"] = "quorum",
                ["x-dead-letter-exchange"] = "bookstore.dlx",
                ["x-dead-letter-routing-key"] = "billing.order-placed"
            });
        await channel.QueueBindAsync("billing.order-placed", "bookstore.ordering", "order.placed");

        await channel.QueueDeclareAsync("billing.order-placed.dlq", durable: true, exclusive: false, autoDelete: false,
            arguments: new Dictionary<string, object?> { ["x-queue-type"] = "quorum" });
        await channel.QueueBindAsync("billing.order-placed.dlq", "bookstore.dlx", "billing.order-placed");

        // bekleme queue'ları: consumer'ı yok, TTL dolunca mesaj default exchange ile
        // yalnız billing.order-placed'e döner (Notification ikinci kopyayı görmez)
        foreach ((string queue, TimeSpan delay) in Levels)
        {
            await channel.QueueDeclareAsync(queue, durable: true, exclusive: false, autoDelete: false,
                arguments: new Dictionary<string, object?>
                {
                    ["x-queue-type"] = "quorum",
                    ["x-message-ttl"] = (int)delay.TotalMilliseconds,
                    ["x-dead-letter-exchange"] = "",
                    ["x-dead-letter-routing-key"] = "billing.order-placed"
                });
            await channel.QueueBindAsync(queue, RetryExchange, queue);
        }
    }
}

İki tasarım kararı var. Bekleme queue'sundan dönen mesaj bookstore.ordering topic exchange'ine değil, default exchange ("") üzerinden doğrudan billing.order-placed'e gidiyor. Topic exchange'e geri yayınlasaydık aynı routing key'i dinleyen Notification da mesajı ikinci kez alırdı; retry yalnız hata alan consumer'ın meselesi. İkinci karar: her bekleme süresi için ayrı queue ve queue seviyesinde TTL. Tek bir retry queue'suna farklı süreli mesajlar koyup mesaj başına expiration verirsen, RabbitMQ süresi dolan mesajı ancak queue'nun başına geldiğinde dead-letter eder; 10 dakikalık bir mesajın arkasındaki 5 saniyelik mesaj da 10 dakika bekler.

Handler'ın kendisi:

public sealed class OrderPlacedRetryHandler(IChannel channel, FlakyPaymentGateway payments, Stopwatch clock)
{
    private const string RetryCountHeader = "x-retry-count";
    private static readonly JsonSerializerOptions JsonOptions = new(JsonSerializerDefaults.Web);

    public async Task HandleAsync(BasicDeliverEventArgs ea)
    {
        int retryCount = ReadRetryCount(ea.BasicProperties);
        try
        {
            OrderPlaced message = JsonSerializer.Deserialize<OrderPlaced>(ea.Body.Span, JsonOptions)!;
            await payments.ChargeAsync(message);
            await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
            Log(ea, retryCount, "charged, ack");
        }
        catch (Exception ex) when (IsTransient(ex) && retryCount < RetryTopology.Levels.Length)
        {
            (string queue, TimeSpan delay) = RetryTopology.Levels[retryCount];
            var props = new BasicProperties(ea.BasicProperties)
            {
                Headers = new Dictionary<string, object?>(ea.BasicProperties.Headers ?? new Dictionary<string, object?>())
                {
                    [RetryCountHeader] = retryCount + 1
                }
            };
            // önce kopyayı bekleme queue'suna yaz (confirm bekler), sonra orijinali ack'le
            await channel.BasicPublishAsync(RetryTopology.RetryExchange, queue, mandatory: true, props, ea.Body);
            await channel.BasicAckAsync(ea.DeliveryTag, multiple: false);
            Log(ea, retryCount, $"{ex.GetType().Name}, retry in {delay.TotalSeconds}s via {queue}");
        }
        catch (Exception ex)
        {
            // kalıcı hata ya da deneme hakkı bitti: requeue yok, DLX devreye girer
            await channel.BasicRejectAsync(ea.DeliveryTag, requeue: false);
            Log(ea, retryCount, $"{ex.GetType().Name}, dead-lettered");
        }
    }

    // IsTransient yukarıda; ReadRetryCount header'ı int olarak okur, Log tek satır yazar
}

Kanal publisherConfirmationTrackingEnabled: true ile açıldığı için BasicPublishAsync broker kopyayı kabul edene kadar dönmüyor; orijinal ancak ondan sonra ack'leniyor. Arada process çökerse mesaj hem bekleme queue'sunda hem asıl queue'da olur, yani bir kopya fazla işlenir. Bu, at-least-once'ın kodda karşılığı ve handler'ın idempotent olmasını gerektiriyor (serinin onuncu yazısı). Deneme sayısını kendi header'ımızda taşıyoruz. Broker'ın eklediği x-death header'ı daha cazip görünse de 3.13'ten beri x- ile başlayan header'lar broker'a ait sayılıyor ve client'ın yeniden yayınladığı x-death sayacı ilerletmiyor; bunu bir sonraki yazıda gerçek çıktıyla göreceğiz.

Üç mesaj gönderdim: A ilk iki denemede 503 alıyor, B'nin para birimi geçersiz, C'de sağlayıcı hiç düzelmiyor:

  0.0s msg-A attempt 1: HttpRequestException, retry in 1s via billing.order-placed.retry.1
  0.0s msg-B attempt 1: InvalidOperationException, dead-lettered
  0.0s msg-C attempt 1: HttpRequestException, retry in 1s via billing.order-placed.retry.1
  1.0s msg-A attempt 2: HttpRequestException, retry in 2s via billing.order-placed.retry.2
  1.0s msg-C attempt 2: HttpRequestException, retry in 2s via billing.order-placed.retry.2
  3.0s msg-A attempt 3: charged, ack
  3.0s msg-C attempt 3: HttpRequestException, retry in 4s via billing.order-placed.retry.3
  7.0s msg-C attempt 4: HttpRequestException, dead-lettered
 11.0s billing.order-placed.dlq message count: 2

B hiç beklemeden DLQ'ya gitti, A üçüncü denemede başardı, C dört denemenin sonunda 7. saniyede DLQ'ya düştü. Beklemeler sırasında consumer başka mesajlara açıktı; hiçbir prefetch slot'u uyumadı.

RabbitMQ 4.3+: quorum queue'nun kendi delayed retry'ı

RabbitMQ 4.3 ile quorum queue'lara yerleşik bir delayed retry geldi; ek queue, ek exchange ve yeniden yayınlama gerekmiyor. Ayar queue argument'ı (ya da policy) ile yapılıyor:

await channel.QueueDeclareAsync(q, durable: true, exclusive: false, autoDelete: false,
    arguments: new Dictionary<string, object?>
    {
        ["x-queue-type"] = "quorum",
        ["x-delivery-limit"] = 4,
        ["x-delayed-retry-type"] = "failed",
        ["x-delayed-retry-min"] = 1000,
        ["x-delayed-retry-max"] = 30000
    });

failed tipi yalnız delivery-count'u artan, yani reject edilen mesajları geciktirir. Bekleme min(min × delivery-count, max) formülüyle doğrusal büyür. Consumer her teslimde BasicRejectAsync(requeue: true) çağırıyor:

   0.0s delivery x-delivery-count=- -> reject(requeue)
   1.0s delivery x-delivery-count=1 -> reject(requeue)
   3.0s delivery x-delivery-count=2 -> reject(requeue)
   6.0s delivery x-delivery-count=3 -> reject(requeue)
  10.1s delivery x-delivery-count=4 -> reject(requeue)
  15.0s messages left in queue: 0

Beklemeler 1, 2, 3 ve 4 saniye; beşinci teslimden sonra delivery limit (4) aşıldı. Bu demo queue'sunda DLX yoktu, mesaj silindi; gerçek queue'da DLQ'ya düşer. Bekleyen mesaj arkadakileri tutmuyor: prefetch 1 ile bir zehirli ve üç sağlam mesajla denediğimde sağlam üç mesaj aynı saniye içinde işlendi, zehirli olan 1. ve 3. saniyede geri geldi. Kısıtları da net: yalnız quorum queue'da, yalnız 4.3 ve sonrasında çalışıyor ve eğri üstel değil doğrusal. 4.4 bunun yanına mesaj bazında zamanlama ekledi: x-opt-delivery-time header'ına Unix epoch milisaniye cinsinden bir zaman yazarsan mesaj hemen queue'ya girer ama consumer'a ancak o zaman gelir. Bu, TTL queue'ları olmadan kendi üstel eğrini kurmanın yolu:

long due = DateTimeOffset.UtcNow.AddSeconds(4).ToUnixTimeMilliseconds();
var props = new BasicProperties
{
    MessageId = "m-delayed",
    Persistent = true,
    Headers = new Dictionary<string, object?> { ["x-opt-delivery-time"] = due }
};
await channel.BasicPublishAsync("", q, mandatory: true, props, "{}"u8.ToArray());

   0.0s published with x-opt-delivery-time = now + 4s
   4.0s received m-delayed

Bir not: eski "delayed message exchange" plugin'i artık bakımı yapılmayan, arşivlenmiş bir proje ve gecikmiş mesajları tek node'da, replikasız tutuyor. Yeni bir sistemde ona yaslanma.

Azure Service Bus: Abandon, MaxDeliveryCount ve Scheduled Retry

Service Bus'ta "geri bırak" işleminin adı abandon. Lock serbest kalır, DeliveryCount bir artar ve mesaj hemen tekrar teslim edilir. Emulator'deki orders queue'sunda MaxDeliveryCount 3 (Azure'da varsayılan 10):

  2.6s msg-A DeliveryCount=1 -> abandon
  2.7s msg-A DeliveryCount=2 -> abandon
  2.7s msg-A DeliveryCount=3 -> abandon
DLQ msg-A DeliveryCount=3 DeadLetterSource=-
    DeadLetterReason=MaxDeliveryCountExceeded
    DeadLetterErrorDescription=Message could not be consumed after 3 delivery attempts.

Üç teslim 100 milisaniyede bitti ve mesaj DLQ'ya düştü. RabbitMQ'daki nack döngüsünün sınırlı hâli: sayaç seni sonsuzluktan koruyor ama araya bekleme koymuyor. Bekleme için mesajın bir kopyasını ileri bir zamana planlayıp orijinali tamamlıyoruz:

processor.ProcessMessageAsync += async args =>
{
    ServiceBusReceivedMessage message = args.Message;
    int attempt = message.ApplicationProperties.TryGetValue("retry-attempt", out object? value) ? (int)value : 0;
    if (attempt >= 2)
    {
        Console.WriteLine($"{T()} {message.MessageId} attempt {attempt + 1}: charged, complete");
        await args.CompleteMessageAsync(message);
        return;
    }
    // geçici hata: kopyayı ileri bir zamana planla, orijinali tamamla
    var retry = new ServiceBusMessage(message);
    retry.ApplicationProperties["retry-attempt"] = attempt + 1;
    TimeSpan delay = TimeSpan.FromSeconds(2 * Math.Pow(2, attempt));
    await sender.ScheduleMessageAsync(retry, DateTimeOffset.UtcNow.Add(delay));
    await args.CompleteMessageAsync(message);
    Console.WriteLine($"{T()} {message.MessageId} attempt {attempt + 1}: 503, scheduled retry in {delay.TotalSeconds}s");
};

  2.2s msg-A attempt 1: 503, scheduled retry in 2s
  4.3s msg-A attempt 2: 503, scheduled retry in 4s
  8.3s msg-A attempt 3: charged, complete

new ServiceBusMessage(received) body'yi, MessageId'yi ve application property'leri kopyalar; DeliveryCount ise yeni mesajda sıfırdan başlar, bu yüzden deneme sayısını kendi property'mizde taşıyoruz. Planlama ile tamamlama iki ayrı çağrı. Arada process çökerse planlanmış kopya da orijinal de yaşar ve mesaj iki kez işlenir; çözüm yine idempotent consumer. Deneme hakkı bitince DeadLetterMessageAsync ile kendi sebep kodunu yazarak DLQ'ya göndermek, MaxDeliveryCount'u beklemekten daha okunur bir iz bırakır.

Kafka: Retry Topic'leri

Kafka'da mesaj başına ack yok; consumer bir offset'i commit eder ve "bu mesajı sonra tekrar ver" diyemez. Partition'daki bir mesajı beklemek, arkasındaki bütün mesajları beklemek demek. Bu yüzden yaygın çözüm, başarısız mesajı ayrı bir retry topic'ine yazıp asıl partition'da ilerlemek. Router, deneme sayısını ve "en erken ne zaman" bilgisini header'larda taşıyor:

public sealed class BillingRetryRouter(
    IProducer<string, byte[]> producer, string retryTopic, string deadLetterTopic, int maxAttempts, TimeSpan baseDelay)
{
    public string Route(ConsumeResult<string, byte[]> result, Exception ex)
    {
        int attempt = ReadInt(result.Message.Headers, "retry-attempt") + 1;
        string target = attempt < maxAttempts ? retryTopic : deadLetterTopic;
        var headers = new Headers();
        foreach (IHeader h in result.Message.Headers)
        {
            if (h.Key is not ("retry-attempt" or "retry-not-before"))
                headers.Add(h.Key, h.GetValueBytes());
        }
        headers.Add("retry-attempt", Encoding.UTF8.GetBytes(attempt.ToString()));
        if (attempt == 1)
        {
            // ilk hata anındaki konum: zincirin sonunda DLT'de "nereden geldi" sorusunu cevaplar
            headers.Add("original-topic-partition-offset", Encoding.UTF8.GetBytes(result.TopicPartitionOffset.ToString()));
        }
        if (target == retryTopic)
        {
            // 2s, 4s, 8s... üstel bekleme; zaman damgası header'da taşınır
            TimeSpan delay = baseDelay * Math.Pow(2, attempt - 1);
            long notBefore = DateTimeOffset.UtcNow.Add(delay).ToUnixTimeMilliseconds();
            headers.Add("retry-not-before", Encoding.UTF8.GetBytes(notBefore.ToString()));
        }
        else
        {
            headers.Add("dlt-exception-type", Encoding.UTF8.GetBytes(ex.GetType().FullName!));
            headers.Add("dlt-exception-message", Encoding.UTF8.GetBytes(ex.Message));
        }
        // aynı key: retry topic'te de aynı siparişin mesajları aynı partition'a düşer
        producer.Produce(target, new Message<string, byte[]>
        {
            Key = result.Message.Key,
            Value = result.Message.Value,
            Headers = headers
        });
        producer.Flush(TimeSpan.FromSeconds(5));
        return target;
    }

    public void WaitUntilDue(Headers headers, CancellationToken token)
    {
        long notBefore = ReadLong(headers, "retry-not-before");
        TimeSpan wait = DateTimeOffset.FromUnixTimeMilliseconds(notBefore) - DateTimeOffset.UtcNow;
        if (notBefore > 0 && wait > TimeSpan.Zero)
            token.WaitHandle.WaitOne(wait); // Consume döngüsü senkron: thread'i iptal edilebilir şekilde beklet
    }

    // ReadInt/ReadLong: header'daki UTF-8 sayıyı okur, yoksa 0
}

Asıl consumer (billing grubu) ve retry consumer'ı (billing-retry grubu) aynı döngüyü çalıştırıyor: WaitUntilDue, işle, hata olursa Route, sonra her durumda StoreOffset. B siparişinde ödeme hep 503 dönüyor:

  4.0s [billing] msg-A charged
  4.1s [billing] msg-B HttpRequestException -> bookstore.orders.billing.retry
  6.1s [billing-retry] msg-B HttpRequestException -> bookstore.orders.billing.retry
 10.1s [billing-retry] msg-B HttpRequestException -> bookstore.orders.billing.dlt

Asıl partition hiç durmadı; B, 2 ve 4 saniyelik beklemelerden sonra DLT'ye (dead-letter topic) gitti. Bu basit sürümün iki sınırı var. WaitUntilDue consumer thread'ini bloklar; bekleme max.poll.interval.ms'i (varsayılan 300000 ms) aşarsa consumer gruptan atılır ve rebalance olur. Ayrıca farklı sürelerle tek retry topic'i kullanınca 4 saniyelik mesaj arkasındaki 2 saniyelik mesajı bekletir. Uzun beklemelerde süre başına ayrı topic (retry-10s, retry-5m) ve bloklamak yerine Pause/Resume kullanılır. Asıl önemli sınır ise sıralama: B retry topic'indeyken aynı key'e sahip sonraki mesajlar (ör. OrderCancelled) asıl topic'te işlenmeye devam eder. Per-key ordering'e güvenen bir consumer'da retry topic'i, sıralamayı kendi elinle bozmak demek; orada ya partition'ı yerinde bekletirsin ya da o key'in sonraki mesajlarını da park edersin. Kavramın kendisi Messaging serisinin "Message Ordering ve Partitioning" yazısında.

MassTransit: UseMessageRetry ve UseDelayedRedelivery

MassTransit bu katmanları hazır veriyor. UseMessageRetry in-memory retry, UseDelayedRedelivery broker üzerinden bekletme. Billing endpoint'i:

cfg.ReceiveEndpoint("billing-order-placed", e =>
{
    if (mode == "redelivery")
    {
        e.UseDelayedRedelivery(r => r.Intervals(TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(30)));
    }
    e.UseMessageRetry(r =>
    {
        r.Exponential(retryLimit: 3, minInterval: TimeSpan.FromMilliseconds(200),
            maxInterval: TimeSpan.FromSeconds(2), intervalDelta: TimeSpan.FromMilliseconds(300));
        r.Ignore<InvalidOperationException>();
    });
    e.ConfigureConsumer<OrderPlacedConsumer>(context);
});

Consumer her çağrıda context.GetRetryAttempt() ve context.GetRedeliveryCount() değerlerini log'luyor. İki sipariş yayınladım: biri EUR (sağlayıcı ilk üç çağrıda 503 dönüyor), biri XXX:

05:01:29.547 info: Bookstore.Billing.OrderPlacedConsumer[0] Order 7259cf10 retry attempt 0, redelivery 0
05:01:29.547 info: Bookstore.Billing.OrderPlacedConsumer[0] Order 58147d8d retry attempt 0, redelivery 0
05:01:29.559 warn: MassTransit.ReceiveTransport[0] R-RETRY rabbitmq://localhost:17301/billing-order-placed 6c0a0000-0000-0000-379a-08df3d3c3452 MassTransit.RetryPolicies.RetryConsumeContext<Bookstore.Contracts.OrderPlaced> System.Net.Http.HttpRequestException: 503 Service Unavailable from payment provider
05:01:29.576 fail: MassTransit.ReceiveTransport[0] R-FAULT rabbitmq://localhost:17301/billing-order-placed 6c0a0000-0000-0000-43d4-08df3d3c3457 Bookstore.Contracts.OrderPlaced Bookstore.Billing.OrderPlacedConsumer(00:00:00.0049625) System.InvalidOperationException: Unsupported currency XXX
05:01:29.991 info: Bookstore.Billing.OrderPlacedConsumer[0] Order 7259cf10 retry attempt 1, redelivery 0
05:01:31.072 info: Bookstore.Billing.OrderPlacedConsumer[0] Order 7259cf10 retry attempt 2, redelivery 0
05:01:32.386 info: Bookstore.Billing.OrderPlacedConsumer[0] Order 7259cf10 retry attempt 3, redelivery 0

(Log satırlarındaki stack trace'leri kısalttım.) XXX siparişi Ignore sayesinde hiç retry edilmeden R-FAULT ile _error queue'suna gitti; EUR siparişi üçüncü retry'da (dördüncü çağrıda) başardı. Bütün bu süre boyunca mesaj consumer'da unacked bekledi; yani bu da in-memory retry ve yukarıdaki maliyet kuralı burada da geçerli.

Daha uzun beklemeler için UseDelayedRedelivery var ve sırası önemli: dokümandaki örnekte olduğu gibi önce redelivery, sonra retry tanımlanır; her redelivery turunda in-memory retry baştan çalışır. Ama MassTransit 8, RabbitMQ'da bu iş için delayed exchange plugin'ine ihtiyaç duyuyor. Plugin'siz bir RabbitMQ 4.4'te aynı kodu çalıştırınca:

05:01:42.574 fail: MassTransit.ReceiveTransport[0] S-FAULT rabbitmq://localhost:17301/billing-order-placed d20a0000-0000-0000-d22e-08df3d3c3c16 Bookstore.Contracts.OrderPlaced RabbitMQ.Client.Exceptions.OperationInterruptedException: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text='PRECONDITION_FAILED - unknown exchange type 'x-delayed-message'', classId=40

Redelivery'nin hata yolu kendisi patladı. Bir detay daha: Ignore<InvalidOperationException> yalnız retry filtresinde tanımlı olduğu için XXX siparişi de redelivery'ye girmeye çalıştı; redelivery'nin kendi filtresi ayrıca yazılır. MassTransit 9.1 ile RabbitMQ için TTL + DLX kullanan, plugin istemeyen UseQueueBasedDelayedRedelivery geldi; v9 ticari lisanslı ve lisans tarafını serinin son yazısında ele alıyoruz. v8'de kalıyorsan uzun beklemeleri yukarıdaki kendi TTL + DLX topology'n ya da quorum queue'nun delayed retry'ı ile çözmek en temiz yol.

Üretimde Dikkat

Retry'ı ölç. Deneme sayısı, retry'a giden mesaj oranı ve DLQ'ya düşme oranı ayrı metric'ler olmalı; retry oranındaki ani artış, sağlayıcının çöktüğünü DLQ alarmından dakikalar önce söyler. Bunları OpenTelemetry ile toplamayı serinin on ikinci yazısında kuracağız.

Toplam retry penceresini iş kuralına bağla. 1 + 2 + 4 saniye bir demo için iyi; gerçek sistemde "ödeme 30 dakika içinde alınamazsa siparişi iptal et" gibi bir kural varsa retry zinciri bunun içinde bitmeli. Mesajın kendi TTL'i retry penceresinden kısaysa mesaj, retry queue'sunda beklerken sessizce sona erer.

Herkes aynı anda geri dönmesin. Sağlayıcı 10 dakika çöktüğünde binlerce mesaj aynı TTL'le bekleme queue'suna girer ve sağlayıcı ayağa kalktığı saniye hepsi birden geri döner. Bekleme sürelerine küçük bir rastgelelik (jitter) eklemek ya da consumer tarafında eşzamanlılığı sınırlamak bu dalgayı yayar. Sabit TTL'li queue'larda jitter yoktur; 4.4'ün x-opt-delivery-time'ı ile mesaj başına rastgele zaman verebilirsin.

Retry, idempotency ister. Retry eden her yapı aynı mesajı en az bir kez fazladan teslim edebilir: confirm ile ack arası, schedule ile complete arası, Kafka'da produce ile offset commit arası. Ödeme gibi yan etkisi olan işlerde sağlayıcıya bir idempotency key (ör. OrderId) gönder ve consumer'ı serinin onuncu yazısındaki gibi kur.

Bekleme queue'larına sınır koy. Retry queue'ları da bellek ve disk tüketir. x-max-length ve overflow ayarı olmayan bir bekleme queue'su, uzun bir kesintide broker'ın disk alarmını tetikleyebilir.

Sık Düşülen Tuzaklar

BasicNackAsync(requeue: true) ile retry. Yukarıda gördüğün gibi 5 saniyede 8036 teslim ve 4.3+ quorum queue'da delivery limit'e takılmayan bir döngü. Requeue'yu yalnız "bu consumer kapanıyor, mesajı başkası alsın" gibi durumlarda kullan.

DLX'siz delivery limit. Reject'le sınırlı bir döngü kurdun, güzel; ama DLX yoksa 21. teslimde mesaj silinir ve kimse fark etmez.

Kalıcı hatayı retry etmek. Bozuk JSON'u dört kez denemek hem gecikme hem gürültü üretir ve gerçek geçici hataları bulmayı zorlaştırır. Exception sınıflandırmasını açıkça yaz.

Idempotent olmayan handler'da retry. Ödemeyi aldıktan sonra fatura kaydında hata alan bir handler'ı retry etmek, müşteriden iki kez para çekmek demek. Önce yan etkisi tekrar edilebilir hale getir, sonra retry ekle.

Retry kopyasını topic exchange'e yayınlamak. Mesaj aynı routing key'le geri gelir ve hata almamış Notification da onu ikinci kez işler. Retry'ı yalnız hata alan queue'ya yönlendir.

Tek retry queue'sunda farklı süreler. Mesaj başına expiration ile kurulan tek queue'da kısa süreli mesajlar uzun süreli mesajın arkasında bekler.

Kafka'da retry topic'inin sıralamayı bozduğunu unutmak. Aynı siparişin iptal mesajı, retry'daki oluşturma mesajından önce işlenebilir.

Küçük Bir Deneme

★☆☆ Requeue döngüsü deneyindeki queue'ya x-dead-letter-exchange ekle ve BasicRejectAsync(requeue: true) ile tekrar çalıştır. Mesajın 21. teslimden sonra DLQ'ya düştüğünü ve x-first-death-reason header'ında hangi değerin yazdığını gör.

★★☆ TTL + DLX topology'sini kaldırıp aynı handler'ı quorum queue'nun x-delayed-retry-type ayarıyla yeniden yaz. Hangi kod satırları gitti? Bekleme eğrisi nasıl değişti ve senin senaryon için doğrusal bekleme yeterli mi?

★★★ Kafka örneğinde aynı key ile önce OrderPlaced, hemen ardından OrderCancelled üret ve ilkini bir kez başarısız yap. İşlenme sırasını log'la. Sonra sıralamayı koruyan bir çözüm kur: aynı key'in mesajları retry'dayken asıl topic'teki yeni mesajları da retry zincirine park et.

Retry ne kadar iyi kurulursa kurulsun, bazı mesajlar yolun sonunda DLQ'ya düşecek: B ve C gibi. Bir sonraki yazıda o noktadan devam ediyoruz: DLQ'yu izlemek, mesajın neden düştüğünü okumak ve düzelttikten sonra güvenle yeniden işlemek.