Request Timeout

Every RPC request requires a timeout in milliseconds. If no response arrives before the timeout expires, the sender receives error code 301 (Request Timeout).

The diagram below shows both outcomes for the same request: a response that arrives in time, and one where the timeout fires before the responder replies.

The sender unblocks with error 301 the moment the timeout elapses — the responder may finish later, but its reply is discarded.

Setting Timeout per Request

timeout.go

resp, err := client.SendCommand(ctx, kubemq.NewCommand().
    SetChannel("orders.process").
    SetBody([]byte("create order")).
    SetTimeout(5 * time.Second))

timeout.py

response = client.send_command(
    CommandMessage(
        channel="orders.process",
        body=b"create order",
        timeout_in_seconds=5,
    )
)

timeout.js

const response = await client.sendCommand({
  channel: "orders.process",
  body: Buffer.from("create order"),
  timeoutInSeconds: 5,
});

Timeout.java

CommandResponseMessage response = client.sendCommandRequest(
    CommandMessage.builder()
        .channel("orders.process")
        .body("create order".getBytes())
        .timeout(5000) // milliseconds
        .build());

Timeout.cs

var response = await client.SendCommandAsync(new CommandMessage
{
    Channel = "orders.process",
    Body = Encoding.UTF8.GetBytes("create order"),
    Timeout = TimeSpan.FromSeconds(5)
});

Timeout.kt

val response = client.sendCommand(CommandMessage(
    channel = "orders.process",
    body = "create order".toByteArray(),
    timeout = 5000 // milliseconds
))

timeout.cpp

kubemq::CommandMessage cmd;
cmd.channel = "orders.process";
cmd.body = "create order";
cmd.timeout = 5000; // milliseconds

auto response = client.sendCommand(cmd);

timeout.rs

use kubemq::prelude::*;
use kubemq::CommandBuilder;
use std::time::Duration;

let command = CommandBuilder::new()
    .channel("orders.process")
    .body(b"create order".to_vec())
    .timeout(Duration::from_secs(5))
    .build();

match client.send_command(command).await {
    Ok(resp) => println!("executed={}, error='{}'", resp.executed, resp.error),
    Err(e) => println!("request timed out: {}", e),
}

timeout.rb

msg = KubeMQ::CQ::CommandMessage.new(
  channel: "orders.process",
  body: "create order",
  timeout: 5 # seconds
)

result = client.send_command(msg)
puts "executed=#{result.executed}, error=#{result.error}"

timeout.exs

cmd =
  KubeMQ.Command.new(
    channel: "orders.process",
    body: "create order",
    timeout: 5_000 # milliseconds
  )

case KubeMQ.Client.send_command(client, cmd) do
  {:ok, response} -> IO.puts("executed: #{response.executed}")
  {:error, err} -> IO.puts("request timed out: #{err.message}")
end

What Happens on Timeout

The sender receives error code 301 (Request Timeout)
The responder may still be processing — KubeMQ does not cancel in-flight work
The request is not automatically retried

Retry Strategies

Simple Retry

Retry a fixed number of times on failure or timeout.

simple_retry.go

func sendWithRetry(ctx context.Context, client *kubemq.Client,
    cmd *kubemq.Command, maxRetries int) (*kubemq.CommandResponse, error) {
    var lastErr error
    for i := 0; i <= maxRetries; i++ {
        resp, err := client.SendCommand(ctx, cmd)
        if err == nil && resp.Executed {
            return resp, nil
        }
        lastErr = err
        if err != nil {
            log.Printf("Attempt %d failed: %v", i+1, err)
        } else {
            log.Printf("Attempt %d failed: %s", i+1, resp.Error)
        }
    }
    return nil, fmt.Errorf("all %d retries failed: %w", maxRetries+1, lastErr)
}

simple_retry.py

def send_with_retry(client, message, max_retries=3):
    last_error = None
    for attempt in range(max_retries + 1):
        try:
            response = client.send_command(message)
            if response.is_executed:
                return response
            last_error = response.error
            print(f"Attempt {attempt + 1} failed: {response.error}")
        except Exception as e:
            last_error = str(e)
            print(f"Attempt {attempt + 1} failed: {e}")
    raise RuntimeError(f"All {max_retries + 1} retries failed: {last_error}")

simple_retry.js

async function sendWithRetry(client, opts, maxRetries = 3) {
  let lastError;
  for (let i = 0; i <= maxRetries; i++) {
    try {
      const response = await client.sendCommand(opts);
      if (response.isExecuted) return response;
      lastError = response.error;
      console.log(`Attempt ${i + 1} failed: ${response.error}`);
    } catch (err) {
      lastError = err.message;
      console.log(`Attempt ${i + 1} failed: ${err.message}`);
    }
  }
  throw new Error(`All ${maxRetries + 1} retries failed: ${lastError}`);
}

SimpleRetry.java

CommandResponseMessage sendWithRetry(CQClient client,
    CommandMessage msg, int maxRetries) throws Exception {
    Exception lastError = null;
    for (int i = 0; i <= maxRetries; i++) {
        try {
            var resp = client.sendCommandRequest(msg);
            if (resp.isExecuted()) return resp;
            System.out.printf("Attempt %d failed: %s%n", i + 1, resp.getError());
        } catch (Exception e) {
            lastError = e;
            System.out.printf("Attempt %d failed: %s%n", i + 1, e.getMessage());
        }
    }
    throw new RuntimeException("All retries failed", lastError);
}

SimpleRetry.cs

async Task<CommandResponse> SendWithRetry(KubeMQClient client,
    CommandMessage msg, int maxRetries = 3)
{
    Exception? lastError = null;
    for (int i = 0; i <= maxRetries; i++)
    {
        try
        {
            var resp = await client.SendCommandAsync(msg);
            if (resp.IsExecuted) return resp;
            Console.WriteLine($"Attempt {i + 1} failed: {resp.Error}");
        }
        catch (Exception ex)
        {
            lastError = ex;
            Console.WriteLine($"Attempt {i + 1} failed: {ex.Message}");
        }
    }
    throw new InvalidOperationException("All retries failed", lastError);
}

SimpleRetry.kt

fun sendWithRetry(client: CQClient, msg: CommandMessage,
    maxRetries: Int = 3): CommandResponse {
    var lastError: Exception? = null
    repeat(maxRetries + 1) { attempt ->
        try {
            val resp = client.sendCommand(msg)
            if (resp.isExecuted) return resp
            println("Attempt ${attempt + 1} failed: ${resp.error}")
        } catch (e: Exception) {
            lastError = e
            println("Attempt ${attempt + 1} failed: ${e.message}")
        }
    }
    throw RuntimeException("All retries failed", lastError)
}

simple_retry.cpp

kubemq::CommandResponse sendWithRetry(kubemq::CQClient& client,
    kubemq::CommandMessage& cmd, int maxRetries = 3) {
    std::string lastError;
    for (int i = 0; i <= maxRetries; i++) {
        try {
            auto resp = client.sendCommand(cmd);
            if (resp.isExecuted) return resp;
            lastError = resp.error;
        } catch (const std::exception& e) {
            lastError = e.what();
        }
    }
    throw std::runtime_error("All retries failed: " + lastError);
}

simple_retry.rs

use kubemq::prelude::*;

async fn send_with_retry(
    client: &KubemqClient,
    command: Command,
    max_retries: u32,
) -> Result<CommandResponse, String> {
    let mut last_error = String::new();
    for attempt in 0..=max_retries {
        match client.send_command(command.clone()).await {
            Ok(resp) if resp.executed => return Ok(resp),
            Ok(resp) => last_error = resp.error,
            Err(e) => last_error = e.to_string(),
        }
        println!("Attempt {} failed: {}", attempt + 1, last_error);
    }
    Err(format!("all {} retries failed: {}", max_retries + 1, last_error))
}

simple_retry.rb

def send_with_retry(client, msg, max_retries = 3)
  last_error = nil
  (0..max_retries).each do |attempt|
    begin
      result = client.send_command(msg)
      return result if result.error.to_s.empty? && result.executed
      last_error = result.error
    rescue KubeMQ::Error => e
      last_error = e.message
    end
    puts "Attempt #{attempt + 1} failed: #{last_error}"
  end
  raise "All #{max_retries + 1} retries failed: #{last_error}"
end

simple_retry.exs

defmodule Retry do
  def send_with_retry(client, cmd, max_retries \\ 3) do
    do_send(client, cmd, 0, max_retries, nil)
  end

  defp do_send(_client, _cmd, attempt, max, last) when attempt > max do
    {:error, "all #{max + 1} retries failed: #{inspect(last)}"}
  end

  defp do_send(client, cmd, attempt, max, _last) do
    case KubeMQ.Client.send_command(client, cmd) do
      {:ok, %{executed: true} = resp} ->
        {:ok, resp}

      {:ok, %{error: error}} ->
        IO.puts("Attempt #{attempt + 1} failed: #{error}")
        do_send(client, cmd, attempt + 1, max, error)

      {:error, err} ->
        IO.puts("Attempt #{attempt + 1} failed: #{err.message}")
        do_send(client, cmd, attempt + 1, max, err.message)
    end
  end
end

Exponential Backoff

Increase the delay between retries to avoid overwhelming a recovering service.

backoff.go

func sendWithBackoff(ctx context.Context, client *kubemq.Client,
    cmd *kubemq.Command, maxRetries int) (*kubemq.CommandResponse, error) {
    for i := 0; i <= maxRetries; i++ {
        resp, err := client.SendCommand(ctx, cmd)
        if err == nil && resp.Executed {
            return resp, nil
        }
        if i < maxRetries {
            delay := time.Duration(1<<uint(i)) * time.Second // 1s, 2s, 4s, 8s...
            log.Printf("Retry in %v...", delay)
            time.Sleep(delay)
        }
    }
    return nil, fmt.Errorf("all retries exhausted")
}

backoff.py

import time

def send_with_backoff(client, message, max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            response = client.send_command(message)
            if response.is_executed:
                return response
        except Exception:
            pass
        if attempt < max_retries:
            delay = 2 ** attempt  # 1s, 2s, 4s, 8s...
            print(f"Retry in {delay}s...")
            time.sleep(delay)
    raise RuntimeError("All retries exhausted")

backoff.js

async function sendWithBackoff(client, opts, maxRetries = 3) {
  for (let i = 0; i <= maxRetries; i++) {
    try {
      const response = await client.sendCommand(opts);
      if (response.isExecuted) return response;
    } catch {}
    if (i < maxRetries) {
      const delay = Math.pow(2, i) * 1000; // 1s, 2s, 4s, 8s...
      console.log(`Retry in ${delay}ms...`);
      await new Promise((r) => setTimeout(r, delay));
    }
  }
  throw new Error("All retries exhausted");
}

Backoff.java

CommandResponseMessage sendWithBackoff(CQClient client,
    CommandMessage msg, int maxRetries) throws Exception {
    for (int i = 0; i <= maxRetries; i++) {
        try {
            var resp = client.sendCommandRequest(msg);
            if (resp.isExecuted()) return resp;
        } catch (Exception ignored) {}
        if (i < maxRetries) {
            long delay = (long) Math.pow(2, i) * 1000;
            System.out.printf("Retry in %dms...%n", delay);
            Thread.sleep(delay);
        }
    }
    throw new RuntimeException("All retries exhausted");
}

Backoff.cs

async Task<CommandResponse> SendWithBackoff(KubeMQClient client,
    CommandMessage msg, int maxRetries = 3)
{
    for (int i = 0; i <= maxRetries; i++)
    {
        try
        {
            var resp = await client.SendCommandAsync(msg);
            if (resp.IsExecuted) return resp;
        }
        catch { }
        if (i < maxRetries)
        {
            var delay = TimeSpan.FromSeconds(Math.Pow(2, i));
            Console.WriteLine($"Retry in {delay}...");
            await Task.Delay(delay);
        }
    }
    throw new InvalidOperationException("All retries exhausted");
}

Backoff.kt

suspend fun sendWithBackoff(client: CQClient, msg: CommandMessage,
    maxRetries: Int = 3): CommandResponse {
    repeat(maxRetries + 1) { attempt ->
        try {
            val resp = client.sendCommand(msg)
            if (resp.isExecuted) return resp
        } catch (_: Exception) {}
        if (attempt < maxRetries) {
            val delay = (1L shl attempt) * 1000
            println("Retry in ${delay}ms...")
            Thread.sleep(delay)
        }
    }
    throw RuntimeException("All retries exhausted")
}

backoff.cpp

kubemq::CommandResponse sendWithBackoff(kubemq::CQClient& client,
    kubemq::CommandMessage& cmd, int maxRetries = 3) {
    for (int i = 0; i <= maxRetries; i++) {
        try {
            auto resp = client.sendCommand(cmd);
            if (resp.isExecuted) return resp;
        } catch (...) {}
        if (i < maxRetries) {
            auto delay = std::chrono::seconds(1 << i);
            std::this_thread::sleep_for(delay);
        }
    }
    throw std::runtime_error("All retries exhausted");
}

backoff.rs

use kubemq::prelude::*;
use std::time::Duration;

async fn send_with_backoff(
    client: &KubemqClient,
    command: Command,
    max_retries: u32,
) -> Result<CommandResponse, String> {
    for attempt in 0..=max_retries {
        if let Ok(resp) = client.send_command(command.clone()).await {
            if resp.executed {
                return Ok(resp);
            }
        }
        if attempt < max_retries {
            let delay = Duration::from_secs(1 << attempt); // 1s, 2s, 4s, 8s...
            println!("Retry in {:?}...", delay);
            tokio::time::sleep(delay).await;
        }
    }
    Err("all retries exhausted".to_string())
}

backoff.rb

def send_with_backoff(client, msg, max_retries = 3)
  (0..max_retries).each do |attempt|
    begin
      result = client.send_command(msg)
      return result if result.error.to_s.empty? && result.executed
    rescue KubeMQ::Error
      # fall through to backoff
    end
    if attempt < max_retries
      delay = 2**attempt # 1s, 2s, 4s, 8s...
      puts "Retry in #{delay}s..."
      sleep(delay)
    end
  end
  raise "All retries exhausted"
end

backoff.exs

defmodule Backoff do
  def send_with_backoff(client, cmd, max_retries \\ 3) do
    do_send(client, cmd, 0, max_retries)
  end

  defp do_send(_client, _cmd, attempt, max) when attempt > max do
    {:error, "all retries exhausted"}
  end

  defp do_send(client, cmd, attempt, max) do
    case KubeMQ.Client.send_command(client, cmd) do
      {:ok, %{executed: true} = resp} ->
        {:ok, resp}

      _ ->
        if attempt < max do
          delay = :math.pow(2, attempt) |> round() # 1s, 2s, 4s, 8s...
          IO.puts("Retry in #{delay}s...")
          Process.sleep(delay * 1_000)
        end

        do_send(client, cmd, attempt + 1, max)
    end
  end
end

Idempotency Considerations

When retrying commands, the responder may receive the same request multiple times. Use the RequestID field to deduplicate:

KubeMQ auto-generates a unique RequestID (NUID) for each request
Set a custom RequestID to enable deduplication on the responder side
The responder should track processed request IDs and skip duplicates

Commands should be idempotent when retries are enabled. Creating an order twice with the same ID should produce the same result, not duplicate orders.

Best Practices

Practice	Recommendation
Timeout value	Set slightly longer than expected processing time
Max retries	2–3 for transient errors, 0 for known permanent failures
Backoff base	Start at 1 second, cap at 30 seconds
Monitoring	Log all timeout errors for alerting
Circuit breaker	Use a circuit breaker for persistent failures

Configure Timeouts & Retries