Overview

When message processing fails, you can implement retry strategies ranging from simple immediate retry to exponential backoff with DLQ fallback. KubeMQ's nack mechanism and DLQ policy provide the building blocks.

The diagram below shows the lifecycle of a message that flows through a backoff retry strategy — each failure either schedules a delayed retry or, once the attempt budget is exhausted, routes the message to a Dead Letter Queue (DLQ).

Message lifecycle: each failed delivery schedules a delayed retry until the attempt budget is exhausted, then routes to the DLQ.

Simple Retry (Nack)

The simplest retry — nack the message so it becomes available again immediately.

simple_retry.go

resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
    Channel:            "orders",
    MaxItems:           1,
    WaitTimeoutSeconds: 5,
})
if err != nil {
    log.Fatal(err)
}
for _, m := range resp.Messages {
    err := processOrder(m.Message.Body)
    if err != nil {
        log.Printf("Processing failed (attempt %d): %v", m.Message.Attributes.ReceiveCount, err)
        resp.NAckAll()
        return
    }
}
resp.AckAll()

simple_retry.py

response = client.receive_queue_messages(
    channel="orders",
    max_messages=1,
    wait_timeout_in_seconds=5,
)
for msg in response.messages:
    try:
        process_order(msg.body)
        msg.ack()
    except Exception as e:
        print(f"Processing failed (attempt {msg.receive_count}): {e}")
        msg.nack()

simple_retry.ts

const messages = await client.receiveQueueMessages({
  channel: 'orders',
  maxMessages: 1,
  waitTimeoutSeconds: 5,
});
for (const msg of messages) {
  try {
    await processOrder(msg.body);
    await msg.ack();
  } catch (err) {
    console.log(`Processing failed (attempt ${msg.receiveCount}):`, err);
    await msg.nack();
  }
}

SimpleRetry.java

ReceiveQueueMessagesResponse response = client.receiveQueueMessages(
    ReceiveQueueMessagesRequest.builder()
        .channel("orders").maxMessages(1).waitTimeoutSeconds(5).build());

for (QueueMessageReceived msg : response.getMessages()) {
    try {
        processOrder(msg.getBody());
        msg.ack();
    } catch (Exception e) {
        System.out.printf("Processing failed (attempt %d): %s%n", msg.getReceiveCount(), e);
        msg.nack();
    }
}

SimpleRetry.cs

var response = await client.ReceiveQueueMessagesAsync(new ReceiveQueueMessagesRequest
{
    Channel = "orders", MaxMessages = 1, WaitTimeoutSeconds = 5,
});
foreach (var msg in response.Messages)
{
    try
    {
        ProcessOrder(msg.Body);
        await msg.AckAsync();
    }
    catch (Exception ex)
    {
        Console.WriteLine($"Processing failed (attempt {msg.ReceiveCount}): {ex.Message}");
        await msg.NAckAsync();
    }
}

SimpleRetry.kt

val response = client.receiveQueueMessages(
    channel = "orders", maxMessages = 1, waitTimeoutSeconds = 5)

for (msg in response.messages) {
    try {
        processOrder(msg.body)
        msg.ack()
    } catch (e: Exception) {
        println("Processing failed (attempt ${msg.receiveCount}): ${e.message}")
        msg.nack()
    }
}

simple_retry.cpp

auto response = client.receiveQueueMessages("orders", 1, 5);
for (const auto& msg : response.messages) {
    try {
        processOrder(msg.body);
        msg.ack();
    } catch (const std::exception& e) {
        std::cerr << "Processing failed (attempt " << msg.receiveCount
                  << "): " << e.what() << std::endl;
        msg.nack();
    }
}

simple_retry.rs

let mut receiver = client.new_queue_downstream_receiver().await?;

let response = receiver
    .poll(PollRequest {
        channel: "orders".to_string(),
        max_items: 1,
        wait_timeout_seconds: 5,
        auto_ack: false,
    })
    .await?;

for msg in &response.messages {
    let attempt = msg.message.attributes.as_ref().map_or(0, |a| a.receive_count);
    match process_order(&msg.message.body) {
        Ok(()) => msg.ack().await?,
        Err(e) => {
            eprintln!("Processing failed (attempt {}): {}", attempt, e);
            // nack returns the message to the queue for immediate redelivery
            msg.nack().await?;
        }
    }
}

simple_retry.rb

receiver = client.create_downstream_receiver
request = KubeMQ::Queues::QueuePollRequest.new(
  channel: 'orders', max_items: 1, wait_timeout: 5)
response = receiver.poll(request)

response.messages.each do |msg|
  begin
    process_order(msg.body)
    msg.ack
  rescue StandardError => e
    attempt = msg.attributes&.receive_count
    puts "Processing failed (attempt #{attempt}): #{e.message}"
    # nack returns the message to the queue for immediate redelivery
    msg.nack
  end
end

simple_retry.exs

{:ok, poll} =
  KubeMQ.Client.poll_queue(client,
    channel: "orders",
    max_items: 1,
    wait_timeout: 5_000
  )

Enum.each(poll.messages, fn msg ->
  attempt = if msg.attributes, do: msg.attributes.receive_count, else: 0

  case process_order(msg.body) do
    :ok ->
      KubeMQ.PollResponse.ack_all(poll)

    {:error, reason} ->
      IO.puts("Processing failed (attempt #{attempt}): #{inspect(reason)}")
      # nack returns the message to the queue for immediate redelivery
      KubeMQ.PollResponse.nack_all(poll)
  end
end)

Exponential Backoff with DLQ Fallback

Combine nack-based retry with a delay requeue for exponential backoff, and a DLQ for final failure.

backoff_retry.go

resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
    Channel:            "orders",
    MaxItems:           1,
    WaitTimeoutSeconds: 5,
})
if err != nil {
    log.Fatal(err)
}

for _, m := range resp.Messages {
    err := processOrder(m.Message.Body)
    if err == nil {
        resp.AckAll()
        return
    }

    attempt := int(m.Message.Attributes.ReceiveCount)
    if attempt >= 5 {
        log.Printf("Max retries exceeded, sending to DLQ")
        resp.ReQueueAll("orders.dlq")
        return
    }

    delay := 1 << attempt // 2, 4, 8, 16 seconds
    log.Printf("Retry %d in %ds", attempt, delay)

    retryMsg := kubemq.NewQueueMessage().
        SetChannel("orders").
        SetBody(m.Message.Body).
        SetMetadata(m.Message.Metadata).
        SetTags(m.Message.Tags).
        SetDelaySeconds(delay)

    client.SendQueueMessage(ctx, retryMsg)
    resp.AckAll()
}

backoff_retry.py

response = client.receive_queue_messages(
    channel="orders", max_messages=1, wait_timeout_in_seconds=5)

for msg in response.messages:
    try:
        process_order(msg.body)
        msg.ack()
    except Exception as e:
        attempt = msg.receive_count

        if attempt >= 5:
            print("Max retries exceeded, sending to DLQ")
            msg.requeue("orders.dlq")
            continue

        delay = 2 ** attempt
        print(f"Retry {attempt} in {delay}s")

        client.send_queue_message(QueueMessage(
            channel="orders",
            body=msg.body,
            metadata=msg.metadata,
            delay_in_seconds=delay,
        ))
        msg.ack()

backoff_retry.ts

const messages = await client.receiveQueueMessages({
  channel: 'orders',
  maxMessages: 1,
  waitTimeoutSeconds: 5,
});

for (const msg of messages) {
  try {
    await processOrder(msg.body);
    await msg.ack();
  } catch (err) {
    const attempt = msg.receiveCount;

    if (attempt >= 5) {
      console.log('Max retries exceeded, sending to DLQ');
      await msg.requeue('orders.dlq');
      continue;
    }

    const delay = Math.pow(2, attempt);
    console.log(`Retry ${attempt} in ${delay}s`);

    await client.sendQueueMessage(
      createQueueMessage({
        channel: 'orders',
        body: msg.body,
        metadata: msg.metadata,
        policy: { delaySeconds: delay },
      }),
    );
    await msg.ack();
  }
}

BackoffRetry.java

ReceiveQueueMessagesResponse response = client.receiveQueueMessages(
    ReceiveQueueMessagesRequest.builder()
        .channel("orders").maxMessages(1).waitTimeoutSeconds(5).build());

for (QueueMessageReceived msg : response.getMessages()) {
    try {
        processOrder(msg.getBody());
        msg.ack();
    } catch (Exception e) {
        int attempt = msg.getReceiveCount();

        if (attempt >= 5) {
            System.out.println("Max retries exceeded, sending to DLQ");
            msg.requeue("orders.dlq");
            continue;
        }

        int delay = (int) Math.pow(2, attempt);
        System.out.printf("Retry %d in %ds%n", attempt, delay);

        client.sendQueueMessage(QueueMessage.builder()
            .channel("orders").body(msg.getBody())
            .delaySeconds(delay).build());
        msg.ack();
    }
}

BackoffRetry.cs

var response = await client.ReceiveQueueMessagesAsync(new ReceiveQueueMessagesRequest
{
    Channel = "orders", MaxMessages = 1, WaitTimeoutSeconds = 5,
});

foreach (var msg in response.Messages)
{
    try
    {
        ProcessOrder(msg.Body);
        await msg.AckAsync();
    }
    catch (Exception)
    {
        var attempt = msg.ReceiveCount;

        if (attempt >= 5)
        {
            Console.WriteLine("Max retries exceeded, sending to DLQ");
            await msg.ReQueueAsync("orders.dlq");
            continue;
        }

        var delay = (int)Math.Pow(2, attempt);
        Console.WriteLine($"Retry {attempt} in {delay}s");

        await client.SendQueueMessageAsync(new QueueMessage
        {
            Channel = "orders",
            Body = msg.Body.ToArray(),
            DelaySeconds = delay
        });
        await msg.AckAsync();
    }
}

BackoffRetry.kt

val response = client.receiveQueueMessages(
    channel = "orders", maxMessages = 1, waitTimeoutSeconds = 5)

for (msg in response.messages) {
    try {
        processOrder(msg.body)
        msg.ack()
    } catch (e: Exception) {
        val attempt = msg.receiveCount

        if (attempt >= 5) {
            println("Max retries exceeded, sending to DLQ")
            msg.requeue("orders.dlq")
            continue
        }

        val delay = 2.0.pow(attempt.toDouble()).toInt()
        println("Retry $attempt in ${delay}s")

        client.sendQueueMessage(QueueMessage(
            channel = "orders", body = msg.body, delaySeconds = delay))
        msg.ack()
    }
}

backoff_retry.cpp

auto response = client.receiveQueueMessages("orders", 1, 5);
for (const auto& msg : response.messages) {
    try {
        processOrder(msg.body);
        msg.ack();
    } catch (const std::exception& e) {
        int attempt = msg.receiveCount;

        if (attempt >= 5) {
            std::cout << "Max retries exceeded, sending to DLQ" << std::endl;
            msg.requeue("orders.dlq");
            continue;
        }

        int delay = static_cast<int>(std::pow(2, attempt));
        std::cout << "Retry " << attempt << " in " << delay << "s" << std::endl;

        kubemq::QueueMessage retryMsg;
        retryMsg.channel = "orders";
        retryMsg.body = msg.body;
        retryMsg.delaySeconds = delay;
        client.sendQueueMessage(retryMsg);
        msg.ack();
    }
}

backoff_retry.rs

let mut receiver = client.new_queue_downstream_receiver().await?;

let response = receiver
    .poll(PollRequest {
        channel: "orders".to_string(),
        max_items: 1,
        wait_timeout_seconds: 5,
        auto_ack: false,
    })
    .await?;

for msg in &response.messages {
    if process_order(&msg.message.body).is_ok() {
        msg.ack().await?;
        continue;
    }

    let attempt = msg.message.attributes.as_ref().map_or(0, |a| a.receive_count);
    if attempt >= 5 {
        println!("Max retries exceeded, sending to DLQ");
        msg.re_queue("orders.dlq").await?;
        continue;
    }

    let delay = 1 << attempt; // 2, 4, 8, 16 seconds
    println!("Retry {} in {}s", attempt, delay);

    let retry_msg = QueueMessageBuilder::new()
        .channel("orders")
        .body(msg.message.body.clone())
        .metadata(&msg.message.metadata)
        .delay_seconds(delay)
        .build();

    client.send_queue_message(retry_msg).await?;
    msg.ack().await?;
}

backoff_retry.rb

receiver = client.create_downstream_receiver
request = KubeMQ::Queues::QueuePollRequest.new(
  channel: 'orders', max_items: 1, wait_timeout: 5)
response = receiver.poll(request)

response.messages.each do |msg|
  begin
    process_order(msg.body)
    msg.ack
  rescue StandardError
    attempt = msg.attributes&.receive_count || 0

    if attempt >= 5
      puts 'Max retries exceeded, sending to DLQ'
      # Re-send the message to the DLQ channel, then ack the original.
      client.send_queue_message(KubeMQ::Queues::QueueMessage.new(
        channel: 'orders.dlq', metadata: msg.metadata, body: msg.body))
      msg.ack
      next
    end

    delay = 2**attempt
    puts "Retry #{attempt} in #{delay}s"

    policy = KubeMQ::Queues::QueueMessagePolicy.new(delay_seconds: delay)
    client.send_queue_message(KubeMQ::Queues::QueueMessage.new(
      channel: 'orders', metadata: msg.metadata, body: msg.body, policy: policy))
    msg.ack
  end
end

backoff_retry.exs

{:ok, poll} =
  KubeMQ.Client.poll_queue(client,
    channel: "orders",
    max_items: 1,
    wait_timeout: 5_000
  )

Enum.each(poll.messages, fn msg ->
  case process_order(msg.body) do
    :ok ->
      KubeMQ.PollResponse.ack_all(poll)

    {:error, _reason} ->
      attempt = if msg.attributes, do: msg.attributes.receive_count, else: 0

      if attempt >= 5 do
        IO.puts("Max retries exceeded, sending to DLQ")
        KubeMQ.PollResponse.requeue_all(poll, "orders.dlq")
      else
        delay = Bitwise.bsl(1, attempt) # 2, 4, 8, 16 seconds
        IO.puts("Retry #{attempt} in #{delay}s")

        retry_msg =
          KubeMQ.QueueMessage.new(
            channel: "orders",
            metadata: msg.metadata,
            body: msg.body,
            policy: KubeMQ.QueuePolicy.new(delay_seconds: delay)
          )

        KubeMQ.Client.send_queue_message(client, retry_msg)
        KubeMQ.PollResponse.ack_all(poll)
      end
  end
end)

Retry Strategy Comparison

Strategy	Latency	Load	Complexity	Best For
Immediate nack	Instant retry	High (tight loop)	Low	Transient errors
Fixed delay requeue	Constant wait	Moderate	Medium	Rate-limited APIs
Exponential backoff	Increasing wait	Low	Medium	External service calls
Backoff + DLQ	Increasing + final stop	Lowest	Higher	Production workloads

Retry with Backoff

Overview

Simple Retry (Nack)

Exponential Backoff with DLQ Fallback

Retry Strategy Comparison

Next Steps

Dead Letter Queue

Webhook Delivery

On this page