# Order Processing Pipeline (/learn/queues/scenarios/order-processing)



## Architecture [#architecture]

An e-commerce platform submits orders to a queue. Worker processes pull orders, validate payment, update inventory, and send confirmations. Failed orders are retried with backoff, and permanently failed orders land in a dead letter queue for manual review.

<Mermaid
  chart="graph LR
    API[&#x22;Order API&#x22;]
    OQ{{&#x22;orders queue&#x22;}}
    W1[&#x22;Worker 1&#x22;]
    W2[&#x22;Worker 2&#x22;]
    ACK[&#x22;Ack + Confirm&#x22;]
    DLQ{{&#x22;orders.dlq&#x22;}}
    MON[&#x22;DLQ Monitor&#x22;]

    API -- submit --> OQ
    OQ -- process --> W1
    OQ -- process --> W2
    W1 -- success --> ACK
    W1 -. &#x22;nack / retry&#x22; .-> OQ
    W1 -. &#x22;max retries&#x22; .-> DLQ
    DLQ -- review --> MON

    class OQ,DLQ queue
    class API,W1,W2,ACK,MON client"
/>

*Orders flow through a work queue to competing workers; failures retry, then dead-letter for review.*

## Order Submission [#order-submission]

The API service creates an order and places it on the queue with DLQ policy and a 5-minute expiration.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="order_api.go"
    type Order struct {
        OrderID   string  `json:"orderId"`
        Customer  string  `json:"customer"`
        Items     int     `json:"items"`
        Total     float64 `json:"total"`
        CreatedAt int64   `json:"createdAt"`
    }

    func submitOrder(ctx context.Context, client *kubemq.Client, order Order) error {
        body, _ := json.Marshal(order)
        msg := kubemq.NewQueueMessage().
            SetChannel("orders").
            SetBody(body).
            SetMetadata("order.created").
            SetTags(map[string]string{"customer": order.Customer}).
            SetMaxReceiveCount(5).
            SetMaxReceiveQueue("orders.dlq").
            SetExpirationSeconds(300)

        result, err := client.SendQueueMessage(ctx, msg)
        if err != nil {
            return err
        }
        log.Printf("Order %s submitted: id=%s", order.OrderID, result.MessageID)
        return nil
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="order_api.py"
    import json
    import time

    def submit_order(client, order):
        result = client.send_queue_message(
            QueueMessage(
                channel="orders",
                body=json.dumps(order).encode(),
                metadata="order.created",
                tags={"customer": order["customer"]},
                max_receive_count=5,
                max_receive_queue="orders.dlq",
                expiration_in_seconds=300,
            )
        )
        print(f"Order {order['orderId']} submitted: id={result.id}")
        return result

    submit_order(client, {
        "orderId": "ORD-2001",
        "customer": "alice@example.com",
        "items": 3,
        "total": 149.97,
        "createdAt": int(time.time() * 1000),
    })
    ```
  </Tab>

  <Tab value="Node.js">
    ```typescript title="order_api.ts"
    interface Order {
      orderId: string;
      customer: string;
      items: number;
      total: number;
      createdAt: number;
    }

    async function submitOrder(client: KubeMQClient, order: Order) {
      const result = await client.sendQueueMessage(
        createQueueMessage({
          channel: 'orders',
          body: JSON.stringify(order),
          metadata: 'order.created',
          tags: { customer: order.customer },
          policy: {
            maxReceiveCount: 5,
            maxReceiveQueue: 'orders.dlq',
            expirationSeconds: 300,
          },
        }),
      );
      console.log(`Order ${order.orderId} submitted: id=${result.messageId}`);
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java title="OrderApi.java"
    public void submitOrder(QueuesClient client, Order order) throws Exception {
        byte[] body = objectMapper.writeValueAsBytes(order);

        QueueMessage msg = QueueMessage.builder()
            .channel("orders")
            .body(body)
            .metadata("order.created")
            .tags(Map.of("customer", order.getCustomer()))
            .maxReceiveCount(5)
            .maxReceiveQueue("orders.dlq")
            .expirationSeconds(300)
            .build();

        SendQueueMessageResult result = client.sendQueueMessage(msg);
        System.out.printf("Order %s submitted: id=%s%n", order.getOrderId(), result.getMessageId());
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="OrderApi.cs"
    async Task SubmitOrder(QueuesClient client, Order order)
    {
        var body = JsonSerializer.SerializeToUtf8Bytes(order);

        var result = await client.SendQueueMessageAsync(new QueueMessage
        {
            Channel = "orders",
            Body = body,
            Metadata = "order.created",
            Tags = new Dictionary<string, string> { ["customer"] = order.Customer },
            MaxReceiveCount = 5,
            MaxReceiveQueue = "orders.dlq",
            ExpirationSeconds = 300
        });

        Console.WriteLine($"Order {order.OrderId} submitted: id={result.MessageId}");
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="OrderApi.kt"
    suspend fun submitOrder(client: QueuesClient, order: Order) {
        val body = Json.encodeToString(order).toByteArray()

        val result = client.sendQueueMessage(QueueMessage(
            channel = "orders",
            body = body,
            metadata = "order.created",
            tags = mapOf("customer" to order.customer),
            maxReceiveCount = 5,
            maxReceiveQueue = "orders.dlq",
            expirationSeconds = 300
        ))

        println("Order ${order.orderId} submitted: id=${result.messageId}")
    }
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="order_api.cpp"
    void submitOrder(kubemq::QueuesClient& client, const Order& order) {
        kubemq::QueueMessage msg;
        msg.channel = "orders";
        msg.body = order.toJson();
        msg.metadata = "order.created";
        msg.tags = {{"customer", order.customer}};
        msg.maxReceiveCount = 5;
        msg.maxReceiveQueue = "orders.dlq";
        msg.expirationSeconds = 300;

        auto result = client.sendQueueMessage(msg);
        std::cout << "Order " << order.orderId
                  << " submitted: id=" << result.messageId << std::endl;
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="order_api.rs"
    use kubemq::prelude::*;
    use kubemq::QueueMessageBuilder;
    use std::collections::HashMap;

    async fn submit_order(client: &KubemqClient, order: &Order) -> kubemq::Result<()> {
        let body = serde_json::to_vec(order).unwrap();

        let msg = QueueMessageBuilder::new()
            .channel("orders")
            .body(body)
            .metadata("order.created")
            .tags(HashMap::from([("customer".to_string(), order.customer.clone())]))
            .max_receive_count(5)
            .max_receive_queue("orders.dlq")
            .expiration_seconds(300)
            .build();

        let result = client.send_queue_message(msg).await?;
        println!("Order {} submitted: id={}", order.order_id, result.message_id);
        Ok(())
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="order_api.rb"
    require 'kubemq'
    require 'json'

    def submit_order(client, order)
      policy = KubeMQ::Queues::QueueMessagePolicy.new(
        max_receive_count: 5,
        max_receive_queue: 'orders.dlq',
        expiration_seconds: 300
      )
      msg = KubeMQ::Queues::QueueMessage.new(
        channel: 'orders',
        metadata: 'order.created',
        body: order.to_json,
        tags: { 'customer' => order['customer'] },
        policy: policy
      )
      result = client.send_queue_message(msg)
      puts "Order #{order['orderId']} submitted: id=#{result.id}"
    end
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="order_api.exs"
    defp submit_order(client, order) do
      msg =
        KubeMQ.QueueMessage.new(
          channel: "orders",
          metadata: "order.created",
          body: Jason.encode!(order),
          tags: %{"customer" => order.customer},
          policy:
            KubeMQ.QueuePolicy.new(
              max_receive_count: 5,
              max_receive_queue: "orders.dlq",
              expiration_seconds: 300
            )
        )

      {:ok, result} = KubeMQ.Client.send_queue_message(client, msg)
      IO.puts("Order #{order.order_id} submitted: id=#{result.message_id}")
    end
    ```
  </Tab>
</Tabs>

## Order Worker [#order-worker]

Workers poll for orders, process them, and acknowledge on success. Failed orders are nacked for retry.

<Tabs groupId="language" items="['Go', 'Python', 'Node.js', 'Java', 'C#', 'Kotlin', 'C++', 'Rust', 'Ruby', 'Elixir']">
  <Tab value="Go">
    ```go title="order_worker.go"
    func runWorker(ctx context.Context, client *kubemq.Client, workerID string) {
        for {
            resp, err := client.PollQueue(ctx, &kubemq.PollRequest{
                Channel:            "orders",
                MaxItems:           5,
                WaitTimeoutSeconds: 10,
                VisibilitySeconds:  120,
            })
            if err != nil {
                log.Printf("[%s] Poll error: %v", workerID, err)
                time.Sleep(5 * time.Second)
                continue
            }

            for _, m := range resp.Messages {
                var order Order
                json.Unmarshal(m.Message.Body, &order)

                log.Printf("[%s] Processing order %s (attempt %d)",
                    workerID, order.OrderID, m.Message.Attributes.ReceiveCount)

                if err := processOrder(order); err != nil {
                    log.Printf("[%s] Failed: %v", workerID, err)
                    continue
                }
                log.Printf("[%s] Order %s completed", workerID, order.OrderID)
            }
            resp.AckAll()
        }
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python title="order_worker.py"
    import json
    import time

    def run_worker(client, worker_id):
        while True:
            response = client.receive_queue_messages(
                channel="orders",
                max_messages=5,
                wait_timeout_in_seconds=10,
                visibility_seconds=120,
            )

            for msg in response.messages:
                order = json.loads(msg.body.decode("utf-8"))
                print(f"[{worker_id}] Processing order {order['orderId']} "
                      f"(attempt {msg.receive_count})")

                try:
                    process_order(order)
                    msg.ack()
                    print(f"[{worker_id}] Order {order['orderId']} completed")
                except Exception as e:
                    print(f"[{worker_id}] Failed: {e}")
                    msg.nack()

            time.sleep(1)
    ```
  </Tab>

  <Tab value="Node.js">
    ```typescript title="order_worker.ts"
    async function runWorker(client: KubeMQClient, workerId: string) {
      while (true) {
        const messages = await client.receiveQueueMessages({
          channel: 'orders',
          maxMessages: 5,
          waitTimeoutSeconds: 10,
          visibilitySeconds: 120,
        });

        for (const msg of messages) {
          const order = JSON.parse(new TextDecoder().decode(msg.body));
          console.log(
            `[${workerId}] Processing order ${order.orderId} (attempt ${msg.receiveCount})`,
          );

          try {
            await processOrder(order);
            await msg.ack();
            console.log(`[${workerId}] Order ${order.orderId} completed`);
          } catch (err) {
            console.log(`[${workerId}] Failed:`, err);
            await msg.nack();
          }
        }

        await new Promise((r) => setTimeout(r, 1000));
      }
    }
    ```
  </Tab>

  <Tab value="Java">
    ```java title="OrderWorker.java"
    public void runWorker(QueuesClient client, String workerId) {
        while (true) {
            ReceiveQueueMessagesResponse response = client.receiveQueueMessages(
                ReceiveQueueMessagesRequest.builder()
                    .channel("orders").maxMessages(5)
                    .waitTimeoutSeconds(10).visibilitySeconds(120).build());

            for (QueueMessageReceived msg : response.getMessages()) {
                Order order = objectMapper.readValue(msg.getBody(), Order.class);
                System.out.printf("[%s] Processing order %s (attempt %d)%n",
                    workerId, order.getOrderId(), msg.getReceiveCount());

                try {
                    processOrder(order);
                    msg.ack();
                    System.out.printf("[%s] Order %s completed%n", workerId, order.getOrderId());
                } catch (Exception e) {
                    System.out.printf("[%s] Failed: %s%n", workerId, e.getMessage());
                    msg.nack();
                }
            }
            Thread.sleep(1000);
        }
    }
    ```
  </Tab>

  <Tab value="C#">
    ```csharp title="OrderWorker.cs"
    async Task RunWorker(QueuesClient client, string workerId)
    {
        while (true)
        {
            var response = await client.ReceiveQueueMessagesAsync(new ReceiveQueueMessagesRequest
            {
                Channel = "orders", MaxMessages = 5,
                WaitTimeoutSeconds = 10, VisibilitySeconds = 120,
            });

            foreach (var msg in response.Messages)
            {
                var order = JsonSerializer.Deserialize<Order>(msg.Body.Span);
                Console.WriteLine($"[{workerId}] Processing {order.OrderId} (attempt {msg.ReceiveCount})");

                try
                {
                    ProcessOrder(order);
                    await msg.AckAsync();
                    Console.WriteLine($"[{workerId}] Order {order.OrderId} completed");
                }
                catch (Exception ex)
                {
                    Console.WriteLine($"[{workerId}] Failed: {ex.Message}");
                    await msg.NAckAsync();
                }
            }
            await Task.Delay(1000);
        }
    }
    ```
  </Tab>

  <Tab value="Kotlin">
    ```kotlin title="OrderWorker.kt"
    suspend fun runWorker(client: QueuesClient, workerId: String) {
        while (true) {
            val response = client.receiveQueueMessages(
                channel = "orders", maxMessages = 5,
                waitTimeoutSeconds = 10, visibilitySeconds = 120)

            for (msg in response.messages) {
                val order = Json.decodeFromString<Order>(String(msg.body))
                println("[$workerId] Processing ${order.orderId} (attempt ${msg.receiveCount})")

                try {
                    processOrder(order)
                    msg.ack()
                    println("[$workerId] Order ${order.orderId} completed")
                } catch (e: Exception) {
                    println("[$workerId] Failed: ${e.message}")
                    msg.nack()
                }
            }
            delay(1000)
        }
    }
    ```
  </Tab>

  <Tab value="C++">
    ```cpp title="order_worker.cpp"
    void runWorker(kubemq::QueuesClient& client, const std::string& workerId) {
        while (true) {
            auto response = client.receiveQueueMessages("orders", 5, 10, false, 120);

            for (const auto& msg : response.messages) {
                auto order = Order::fromJson(msg.body);
                std::cout << "[" << workerId << "] Processing " << order.orderId
                          << " (attempt " << msg.receiveCount << ")" << std::endl;

                try {
                    processOrder(order);
                    msg.ack();
                    std::cout << "[" << workerId << "] Order " << order.orderId
                              << " completed" << std::endl;
                } catch (const std::exception& e) {
                    std::cerr << "[" << workerId << "] Failed: " << e.what() << std::endl;
                    msg.nack();
                }
            }
            std::this_thread::sleep_for(std::chrono::seconds(1));
        }
    }
    ```
  </Tab>

  <Tab value="Rust">
    ```rust title="order_worker.rs"
    use kubemq::prelude::*;
    use kubemq::PollRequest;
    use std::time::Duration;

    async fn run_worker(client: &KubemqClient, worker_id: &str) -> kubemq::Result<()> {
        let mut receiver = client.new_queue_downstream_receiver().await?;
        loop {
            let poll = PollRequest {
                channel: "orders".to_string(),
                max_items: 5,
                wait_timeout_seconds: 10,
                auto_ack: false,
            };
            let response = receiver.poll(poll).await?;

            for msg in &response.messages {
                let order: Order = serde_json::from_slice(&msg.message.body).unwrap();
                let attempt = msg.message.attributes.as_ref().map(|a| a.receive_count).unwrap_or(0);
                println!("[{}] Processing order {} (attempt {})", worker_id, order.order_id, attempt);

                match process_order(&order) {
                    Ok(_) => {
                        msg.ack().await?;
                        println!("[{}] Order {} completed", worker_id, order.order_id);
                    }
                    Err(e) => {
                        println!("[{}] Failed: {}", worker_id, e);
                        msg.nack().await?;
                    }
                }
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
        }
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby title="order_worker.rb"
    require 'kubemq'
    require 'json'

    def run_worker(client, worker_id)
      receiver = client.create_downstream_receiver
      loop do
        request = KubeMQ::Queues::QueuePollRequest.new(
          channel: 'orders', max_items: 5, wait_timeout: 10
        )
        response = receiver.poll(request)

        response.messages.each do |m|
          order = JSON.parse(m.body)
          puts "[#{worker_id}] Processing order #{order['orderId']} (attempt #{m.attributes.receive_count})"

          begin
            process_order(order)
            m.ack
            puts "[#{worker_id}] Order #{order['orderId']} completed"
          rescue StandardError => e
            puts "[#{worker_id}] Failed: #{e.message}"
            m.nack
          end
        end
        sleep 1
      end
    ensure
      receiver&.close
    end
    ```
  </Tab>

  <Tab value="Elixir">
    ```elixir title="order_worker.exs"
    defp run_worker(client, worker_id) do
      case KubeMQ.Client.poll_queue(client, channel: "orders", max_items: 5, wait_timeout: 10_000) do
        {:ok, poll} ->
          # The Elixir SDK settles a poll batch as a whole: ack_all on success,
          # nack_all to return every message for redelivery (no per-message ack).
          results =
            Enum.map(poll.messages, fn msg ->
              order = Jason.decode!(msg.body)
              IO.puts("[#{worker_id}] Processing order #{order["orderId"]} (attempt #{msg.attributes.receive_count})")
              process_order(order)
            end)

          if Enum.all?(results, &(&1 == :ok)) do
            KubeMQ.PollResponse.ack_all(poll)
            IO.puts("[#{worker_id}] Batch completed")
          else
            KubeMQ.PollResponse.nack_all(poll)
            IO.puts("[#{worker_id}] Batch returned for retry")
          end

        {:error, err} ->
          IO.puts("[#{worker_id}] Poll error: #{err.message}")
      end

      Process.sleep(1_000)
      run_worker(client, worker_id)
    end
    ```
  </Tab>
</Tabs>

## Advanced Configuration [#advanced-configuration]

<Accordions>
  <Accordion title="Idempotency">
    Use the `orderId` as an idempotency key. Before processing, check if the order was already processed. This prevents duplicate work when messages are redelivered.
  </Accordion>

  <Accordion title="Visibility Tuning">
    Set visibility timeout based on your processing time. For payment processing (5-30s), use 120s. For shipping label generation (30-60s), use 300s. Consider using [extend visibility](/learn/queues/how-to/extend-visibility) for variable-duration tasks.
  </Accordion>

  <Accordion title="DLQ Monitoring">
    Poll the DLQ channel periodically to detect failed orders. Alert on DLQ message count exceeding a threshold. Inspect `reRoutedFromQueue` to identify the source queue.
  </Accordion>
</Accordions>

## Next Steps [#next-steps]

<Cards>
  <Card title="Background Workers" href="/learn/queues/scenarios/background-workers" description="Worker pool pattern for background jobs." />

  <Card title="Retry with Backoff" href="/learn/queues/how-to/retry-with-backoff" description="Exponential backoff strategies." />
</Cards>