# RPC (Direct Reply-To) (/connectors/rabbitmq/how-to/rpc)



**RPC** turns messaging into request/response: a client sends a request and blocks for a reply. AMQP RPC on this connector is **fully native and in-protocol** — there is **no gRPC responder** anywhere. The responder is just a normal AMQP consumer of a request queue that publishes a reply. The recommended mechanism is RabbitMQ's **Direct Reply-To** (`amq.rabbitmq.reply-to`), a pseudo-queue that avoids declaring a real reply queue per request. Every queue involved — the request queue and the reply path — resolves onto ordinary KubeMQ Queue channels.

## Overview [#overview]

The **requester** consumes the `amq.rabbitmq.reply-to` pseudo-queue (it declares no queue) and publishes each request to a request queue, carrying `reply-to = amq.rabbitmq.reply-to` and a unique `correlation-id`. The connector mints an opaque address (`amq.rabbitmq.reply-to.g1.{node}.{id}`) and rewrites the `reply-to` so the responder only ever sees the minted address. The **responder** is a normal consumer of the request queue; it publishes its reply to the default exchange keyed by that minted address, echoing the **same** `correlation-id`. The requester matches each reply to its request by `correlation-id`.

| Operation                  | AMQP action                                                                  | KubeMQ mapping                                 |
| -------------------------- | ---------------------------------------------------------------------------- | ---------------------------------------------- |
| Requester consumes replies | `basic.consume("amq.rabbitmq.reply-to", no-ack=true)`                        | Pseudo-queue (no real channel declared)        |
| Requester sends request    | `basic.publish(routing-key="rpc-queue", reply-to=..., correlation-id=...)`   | `SendQueueMessage` to `amqp.default.rpc-queue` |
| Responder consumes         | `basic.consume("rpc-queue")`                                                 | Competing-consumer pull on the request queue   |
| Responder replies          | `basic.publish(routing-key=req.reply-to, correlation-id=req.correlation-id)` | Reply routed to the minted reply address       |
| Match                      | requester correlates by `correlation-id`                                     | Reply paired to its request                    |

## How it works [#how-it-works]

The connector rewrites the requester's `reply-to` to an opaque, node-local minted address before the request reaches the responder. The responder echoes that address and the correlation-id; the requester pairs each reply to the request it issued.

<Mermaid
  chart="`
sequenceDiagram
participant R as Requester
participant C as RabbitMQ connector
participant B as Message Broker
participant S as Responder

R->>C: consume amq.rabbitmq.reply-to (no-ack)
R->>C: publish request<br/>(key=rpc-queue, reply-to=amq.rabbitmq.reply-to, correlation-id=X)
C->>B: SendQueueMessage → amqp.default.rpc-queue<br/>(reply-to rewritten to minted address)
B->>S: deliver request (minted reply-to)
S->>C: publish reply<br/>(key=minted reply-to, correlation-id=X)
C->>B: route reply to the requester's reply address
B-->>R: deliver reply (correlation-id=X)
Note over R: match reply to request by correlation-id
`"
/>

*The requester's `reply-to` is rewritten to a minted address; the responder echoes the correlation-id, and the requester pairs each reply to its request.*

## Request and reply [#request-and-reply]

Each example runs a responder that consumes `rpc-queue` and echoes each request, and a requester that consumes `amq.rabbitmq.reply-to` (no-ack) and issues five correlated requests, matching each response by `correlation-id`. Every client reads the broker endpoint from `KUBEMQ_AMQP_URL` (default `amqp://guest:guest@localhost:5672/`).

<Tabs groupId="language" items="['Go','Python','Java','JavaScript','C#','Ruby','Rust']">
  <Tab value="Go">
    ```go
    package main

    import (
    	"context"
    	"fmt"
    	"log"
    	"os"
    	"strings"
    	"time"

    	amqp "github.com/rabbitmq/amqp091-go"
    )

    const (
    	rpcQueue = "rpc-queue"
    	replyTo  = "amq.rabbitmq.reply-to"
    	calls    = 5
    )

    func amqpURL() string {
    	if v := os.Getenv("KUBEMQ_AMQP_URL"); v != "" {
    		return v
    	}
    	return "amqp://guest:guest@localhost:5672/"
    }

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    	defer cancel()

    	conn, err := amqp.Dial(amqpURL())
    	if err != nil {
    		log.Fatalf("dial: %v", err)
    	}
    	defer func() { _ = conn.Close() }()

    	// Responder: consume rpc-queue, reply to the minted reply-to address.
    	serverCh, _ := conn.Channel()
    	if _, err := serverCh.QueueDeclare(rpcQueue, false, false, false, false, nil); err != nil {
    		log.Fatalf("declare rpc-queue: %v", err)
    	}
    	requests, err := serverCh.Consume(rpcQueue, "responder", true, false, false, false, nil)
    	if err != nil {
    		log.Fatalf("responder consume: %v", err)
    	}
    	go func() {
    		for req := range requests {
    			// The requester's reply-to is rewritten to a minted opaque address.
    			if !strings.HasPrefix(req.ReplyTo, "amq.rabbitmq.reply-to.g1.") {
    				log.Printf("warning: responder expected a minted address, got %q", req.ReplyTo)
    			}
    			_ = serverCh.PublishWithContext(ctx, "", req.ReplyTo, false, false, amqp.Publishing{
    				ContentType:   "text/plain",
    				CorrelationId: req.CorrelationId,
    				Body:          append([]byte("echo:"), req.Body...),
    			})
    		}
    	}()

    	// Requester: consume the pseudo-queue with no-ack (manual ack → 406).
    	clientCh, _ := conn.Channel()
    	replies, err := clientCh.Consume(replyTo, "", true, false, false, false, nil)
    	if err != nil {
    		log.Fatalf("consuming %s requires no-ack: %v", replyTo, err)
    	}

    	for i := 1; i <= calls; i++ {
    		corr := fmt.Sprintf("corr-%d", i)
    		body := fmt.Sprintf("request-%d", i)
    		if err := clientCh.PublishWithContext(ctx, "", rpcQueue, false, false, amqp.Publishing{
    			ContentType:   "text/plain",
    			CorrelationId: corr,
    			ReplyTo:       replyTo,
    			Body:          []byte(body),
    		}); err != nil {
    			log.Fatalf("request %d: %v", i, err)
    		}
    		select {
    		case reply := <-replies:
    			if reply.CorrelationId != corr {
    				log.Fatalf("call %d: correlation-id %q != %q", i, reply.CorrelationId, corr)
    			}
    			log.Printf(" [rpc] %s (corr=%s) → %s", body, corr, reply.Body)
    		case <-time.After(30 * time.Second):
    			log.Fatalf("timed out waiting for reply %d", i)
    		}
    	}
    	log.Printf(" [✓] %d RPC calls round-tripped over %s, correlation-id matched", calls, replyTo)

    	if _, err := serverCh.QueueDelete(rpcQueue, false, false, false); err != nil {
    		log.Printf("warning: rpc-queue delete: %v", err)
    	}
    }
    ```
  </Tab>

  <Tab value="Python">
    ```python
    import os

    import pika

    URL = os.environ.get("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/")
    RPC_QUEUE = "rpc-queue"
    REPLY_TO = "amq.rabbitmq.reply-to"
    CALLS = 5


    def main() -> None:
        # Responder: consume requests, echo back to the minted reply address.
        server_conn = pika.BlockingConnection(pika.URLParameters(URL))
        server_ch = server_conn.channel()
        server_ch.queue_declare(queue=RPC_QUEUE, durable=False)

        def on_request(ch, method, props, body):
            # The connector rewrites reply-to to an opaque minted address; the
            # responder only ever sees that, never the pseudo-queue name.
            ch.basic_publish(
                exchange="",
                routing_key=props.reply_to,
                body=b"echo:" + body,
                properties=pika.BasicProperties(content_type="text/plain", correlation_id=props.correlation_id),
            )
            ch.basic_ack(method.delivery_tag)

        server_ch.basic_consume(queue=RPC_QUEUE, on_message_callback=on_request)

        # Requester: consume amq.rabbitmq.reply-to (no-ack is mandatory).
        client_conn = pika.BlockingConnection(pika.URLParameters(URL))
        client_ch = client_conn.channel()
        responses: dict[str, str] = {}
        client_ch.basic_consume(queue=REPLY_TO, on_message_callback=lambda c, m, p, b: responses.__setitem__(p.correlation_id, b.decode()), auto_ack=True)

        print(" [client] issuing RPC requests over amq.rabbitmq.reply-to:")
        for i in range(1, CALLS + 1):
            corr = f"corr-{i}"
            client_ch.basic_publish(
                exchange="",
                routing_key=RPC_QUEUE,
                body=f"request-{i}".encode(),
                properties=pika.BasicProperties(content_type="text/plain", correlation_id=corr, reply_to=REPLY_TO),
            )
            # Pump the responder to handle the request, then the client to receive.
            while corr not in responses:
                server_conn.process_data_events(time_limit=1)
                client_conn.process_data_events(time_limit=1)
            print(f"   request-{i} [{corr}] -> {responses[corr]!r}")

        print(f" [x] all {CALLS} responses correlation-id matched (native AMQP RPC)")
        server_ch.queue_delete(queue=RPC_QUEUE)
        client_conn.close()
        server_conn.close()


    if __name__ == "__main__":
        main()
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import java.nio.charset.StandardCharsets;
    import java.util.concurrent.ConcurrentHashMap;
    import java.util.concurrent.SynchronousQueue;
    import java.util.concurrent.TimeUnit;

    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.DeliverCallback;

    public final class Main {
        private static final String RPC_QUEUE = "rpc-queue";
        private static final String REPLY_TO = "amq.rabbitmq.reply-to";
        private static final int CALLS = 5;

        public static void main(String[] args) throws Exception {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setUri(System.getenv().getOrDefault("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"));
            if (factory.getVirtualHost() == null || factory.getVirtualHost().isEmpty()) {
                factory.setVirtualHost("/");
            }

            try (Connection serverConn = factory.newConnection();
                    Connection clientConn = factory.newConnection()) {
                // Responder: consume requests, reply to the minted address.
                Channel serverCh = serverConn.createChannel();
                serverCh.queueDeclare(RPC_QUEUE, false, false, false, null);
                DeliverCallback onRequest = (tag, req) -> {
                    String replyAddr = req.getProperties().getReplyTo(); // minted address
                    AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                            .contentType("text/plain")
                            .correlationId(req.getProperties().getCorrelationId())
                            .build();
                    byte[] body = ("echo:" + new String(req.getBody(), StandardCharsets.UTF_8)).getBytes(StandardCharsets.UTF_8);
                    serverCh.basicPublish("", replyAddr, replyProps, body);
                };
                serverCh.basicConsume(RPC_QUEUE, true, onRequest, tag -> { });

                // Requester: consume the pseudo-queue (auto-ack), then call.
                Channel clientCh = clientConn.createChannel();
                ConcurrentHashMap<String, SynchronousQueue<String>> pending = new ConcurrentHashMap<>();
                DeliverCallback onReply = (tag, reply) -> {
                    SynchronousQueue<String> slot = pending.get(reply.getProperties().getCorrelationId());
                    if (slot != null) {
                        slot.offer(new String(reply.getBody(), StandardCharsets.UTF_8));
                    }
                };
                clientCh.basicConsume(REPLY_TO, true, onReply, tag -> { });

                for (int i = 1; i <= CALLS; i++) {
                    String corr = "corr-" + i;
                    String request = "request-" + i;
                    SynchronousQueue<String> slot = new SynchronousQueue<>();
                    pending.put(corr, slot);
                    AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                            .contentType("text/plain")
                            .correlationId(corr)
                            .replyTo(REPLY_TO)
                            .build();
                    clientCh.basicPublish("", RPC_QUEUE, props, request.getBytes(StandardCharsets.UTF_8));

                    String reply = slot.poll(30, TimeUnit.SECONDS);
                    pending.remove(corr);
                    if (reply == null) throw new IllegalStateException("timed out on reply " + i);
                    System.out.println("[rpc] " + corr + ": '" + request + "' -> '" + reply + "'");
                }
                System.out.println("[x] " + CALLS + " RPC calls matched by correlation-id over amq.rabbitmq.reply-to");
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import amqp from "amqplib";

    const RPC_QUEUE = "rpc-queue";
    const REPLY_TO = "amq.rabbitmq.reply-to";
    const CALLS = 5;

    function url(): string {
      return process.env["KUBEMQ_AMQP_URL"] ?? "amqp://guest:guest@localhost:5672/";
    }

    async function main(): Promise<void> {
      const connection = await amqp.connect(url());

      // Responder: consume rpc-queue, echo back to the reply-to address.
      const serverCh = await connection.createChannel();
      await serverCh.assertQueue(RPC_QUEUE, { durable: false });
      await serverCh.consume(RPC_QUEUE, (req) => {
        if (req === null) return;
        const replyAddr = req.properties.replyTo; // minted address
        serverCh.publish("", replyAddr, Buffer.concat([Buffer.from("echo:"), req.content]), {
          contentType: "text/plain",
          correlationId: req.properties.correlationId,
        });
      }, { noAck: true });

      // Requester: consume amq.rabbitmq.reply-to with no-ack (manual ack → 406).
      const clientCh = await connection.createChannel();
      const pending = new Map<string, (body: string) => void>();
      await clientCh.consume(REPLY_TO, (reply) => {
        if (reply === null) return;
        const resolve = pending.get(reply.properties.correlationId);
        if (resolve) {
          pending.delete(reply.properties.correlationId);
          resolve(reply.content.toString());
        }
      }, { noAck: true });

      for (let i = 1; i <= CALLS; i++) {
        const corr = `corr-${i}`;
        const body = `request-${i}`;
        const response = new Promise<string>((resolve) => pending.set(corr, resolve));
        clientCh.publish("", RPC_QUEUE, Buffer.from(body), {
          contentType: "text/plain",
          correlationId: corr,
          replyTo: REPLY_TO,
        });
        console.log(`[client] ${body} (${corr}) -> ${await response}`);
      }

      console.log(`[client] completed ${CALLS} RPC round-trips, all correlation-id matched`);
      await serverCh.deleteQueue(RPC_QUEUE);
      await connection.close();
    }

    main().catch((err) => {
      console.error(err);
      process.exit(1);
    });
    ```
  </Tab>

  <Tab value="C#">
    ```csharp
    using System.Collections.Concurrent;
    using System.Text;
    using RabbitMQ.Client;
    using RabbitMQ.Client.Events;

    const string requestQueue = "rpc-queue";
    const string replyToPseudoQueue = "amq.rabbitmq.reply-to";
    const int calls = 5;

    static string Url() =>
        Environment.GetEnvironmentVariable("KUBEMQ_AMQP_URL") is { Length: > 0 } v
            ? v
            : "amqp://guest:guest@localhost:5672/";

    var factory = new ConnectionFactory { Uri = new Uri(Url()) };
    await using var connection = await factory.CreateConnectionAsync("rpc");

    // Responder: consume rpc-queue, echo back to the delivered reply-to.
    await using var serverCh = await connection.CreateChannelAsync();
    await serverCh.QueueDeclareAsync(requestQueue, durable: false, exclusive: false, autoDelete: false);
    var serverConsumer = new AsyncEventingBasicConsumer(serverCh);
    serverConsumer.ReceivedAsync += async (_, ea) =>
    {
        var replyTo = ea.BasicProperties.ReplyTo ?? ""; // minted address
        var replyProps = new BasicProperties
        {
            ContentType = "text/plain",
            CorrelationId = ea.BasicProperties.CorrelationId,
        };
        var replyBody = Encoding.UTF8.GetBytes("echo:" + Encoding.UTF8.GetString(ea.Body.Span));
        await serverCh.BasicPublishAsync("", replyTo, mandatory: false, basicProperties: replyProps, body: replyBody);
    };
    await serverCh.BasicConsumeAsync(requestQueue, autoAck: true, consumer: serverConsumer);

    // Requester: consume the pseudo-queue with no-ack (manual ack → 406).
    await using var clientCh = await connection.CreateChannelAsync();
    var pending = new ConcurrentDictionary<string, TaskCompletionSource<string>>();
    var clientConsumer = new AsyncEventingBasicConsumer(clientCh);
    clientConsumer.ReceivedAsync += (_, ea) =>
    {
        if (pending.TryRemove(ea.BasicProperties.CorrelationId ?? "", out var tcs))
            tcs.TrySetResult(Encoding.UTF8.GetString(ea.Body.Span));
        return Task.CompletedTask;
    };
    await clientCh.BasicConsumeAsync(replyToPseudoQueue, autoAck: true, consumer: clientConsumer);

    try
    {
        for (var i = 1; i <= calls; i++)
        {
            var corr = $"corr-{i}";
            var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
            pending[corr] = tcs;
            var reqProps = new BasicProperties
            {
                ContentType = "text/plain",
                CorrelationId = corr,
                ReplyTo = replyToPseudoQueue,
            };
            var request = $"request-{i}";
            await clientCh.BasicPublishAsync("", requestQueue, mandatory: false,
                basicProperties: reqProps, body: Encoding.UTF8.GetBytes(request));
            Console.WriteLine($"[rpc] {corr}: '{request}' → '{await tcs.Task}'");
        }
        Console.WriteLine($"[x] All {calls} RPC calls correlation-id matched over native amq.rabbitmq.reply-to");
    }
    finally
    {
        await serverCh.QueueDeleteAsync(requestQueue, ifUnused: false, ifEmpty: false);
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # frozen_string_literal: true
    require "bunny"
    require "amq/uri"

    RPC_QUEUE = "rpc-queue"
    REPLY_TO = "amq.rabbitmq.reply-to"
    REQUEST_COUNT = 5

    opts = AMQ::URI.parse(ENV.fetch("KUBEMQ_AMQP_URL", "amqp://guest:guest@localhost:5672/"))
    opts[:vhost] = "/" if opts[:vhost].nil? || opts[:vhost].to_s.empty?
    conn = Bunny.new(opts)
    conn.start

    # Responder: consume rpc-queue, echo back with the correlation-id.
    server_ch = conn.create_channel
    rpc_q = server_ch.queue(RPC_QUEUE, durable: false)
    rpc_q.subscribe(manual_ack: false, block: false) do |_di, props, body|
      server_ch.default_exchange.publish("echo:#{body}", routing_key: props.reply_to, correlation_id: props.correlation_id)
    end

    # Requester: consume amq.rabbitmq.reply-to (no-ack MUST be true) and fire 5 requests.
    client_ch = conn.create_channel
    replies = {}
    mutex = Mutex.new
    cond = ConditionVariable.new
    client_ch.basic_consume(REPLY_TO, "", true, false) do |_di, props, body|
      mutex.synchronize do
        replies[props.correlation_id] = body
        cond.signal
      end
    end

    requests = {}
    REQUEST_COUNT.times do |i|
      cid = "corr-#{i + 1}"
      payload = "request-#{i + 1}"
      requests[cid] = payload
      client_ch.default_exchange.publish(payload, routing_key: RPC_QUEUE, reply_to: REPLY_TO, correlation_id: cid)
      puts " [>] request #{cid} body=#{payload.inspect}"
    end

    deadline = Time.now + 10
    mutex.synchronize do
      cond.wait(mutex, 0.5) while replies.size < REQUEST_COUNT && Time.now < deadline
    end

    requests.keys.sort.each { |cid| puts " [<] response #{cid} -> #{replies[cid].inspect}" }
    puts " [x] All #{REQUEST_COUNT} RPC round-trips matched by correlation-id"

    rpc_q.delete
    conn.close
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use futures_lite::StreamExt;
    use lapin::{
        options::{BasicConsumeOptions, BasicPublishOptions, QueueDeclareOptions, QueueDeleteOptions},
        types::FieldTable,
        BasicProperties, Connection, ConnectionProperties,
    };
    use std::collections::BTreeMap;

    const RPC_QUEUE: &str = "rpc-queue";
    const REPLY_TO: &str = "amq.rabbitmq.reply-to";
    const REQUESTS: usize = 5;

    fn amqp_url() -> String {
        let url = std::env::var("KUBEMQ_AMQP_URL")
            .unwrap_or_else(|_| "amqp://guest:guest@localhost:5672/".into());
        match url.rsplit_once('@').map_or(url.as_str(), |(_, host)| host) {
            host if host.ends_with('/') && !host.to_lowercase().ends_with("/%2f") => format!("{url}%2f"),
            _ => url,
        }
    }

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn std::error::Error>> {
        let url = amqp_url();

        // Responder: consume rpc-queue and reply to each request's reply-to address.
        let responder_conn = Connection::connect(&url, ConnectionProperties::default()).await?;
        let responder_ch = responder_conn.create_channel().await?;
        responder_ch
            .queue_declare(RPC_QUEUE, QueueDeclareOptions::default(), FieldTable::default())
            .await?;
        let mut requests = responder_ch
            .basic_consume(RPC_QUEUE, "responder", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
            .await?;

        let responder = tokio::spawn(async move {
            let mut handled = 0usize;
            while handled < REQUESTS {
                let Some(delivery) = requests.next().await else { break };
                let delivery = delivery?;
                let reply_to = delivery.properties.reply_to().as_ref().map(|s| s.to_string()).unwrap_or_default();
                let corr = delivery.properties.correlation_id().clone();
                let mut body = b"echo:".to_vec();
                body.extend_from_slice(&delivery.data);
                // The responder only ever sees the minted reply-to address.
                responder_ch
                    .basic_publish(
                        "",
                        reply_to.as_str(),
                        BasicPublishOptions::default(),
                        &body,
                        BasicProperties::default().with_correlation_id(corr.unwrap_or_else(|| "".into())),
                    )
                    .await?
                    .await?;
                handled += 1;
            }
            Ok::<_, lapin::Error>(())
        });

        // Requester: consume amq.rabbitmq.reply-to (no-ack is mandatory; manual ack → 406).
        let requester_conn = Connection::connect(&url, ConnectionProperties::default()).await?;
        let requester_ch = requester_conn.create_channel().await?;
        let mut replies = requester_ch
            .basic_consume(REPLY_TO, "requester", BasicConsumeOptions { no_ack: true, ..Default::default() }, FieldTable::default())
            .await?;

        let mut pending: BTreeMap<String, String> = BTreeMap::new();
        for i in 1..=REQUESTS {
            let corr = format!("corr-{i}");
            let body = format!("request-{i}");
            pending.insert(corr.clone(), format!("echo:{body}"));
            requester_ch
                .basic_publish(
                    "",
                    RPC_QUEUE,
                    BasicPublishOptions::default(),
                    body.as_bytes(),
                    BasicProperties::default()
                        .with_reply_to(REPLY_TO.into())
                        .with_correlation_id(corr.as_str().into()),
                )
                .await?
                .await?;
        }

        let mut matched = 0usize;
        while matched < REQUESTS {
            let delivery = replies.next().await.ok_or("reply stream closed")??;
            let corr = delivery.properties.correlation_id().as_ref().map(|s| s.to_string()).unwrap_or_default();
            let body = String::from_utf8_lossy(&delivery.data).into_owned();
            if let Some(want) = pending.remove(&corr) {
                assert_eq!(body, want, "reply must echo the request");
                println!("[requester] matched {corr}: {body}");
                matched += 1;
            }
        }
        println!("[x] All {REQUESTS} RPC request/response pairs matched by correlation-id");

        responder.await??;
        let _ = requester_ch.queue_delete(RPC_QUEUE, QueueDeleteOptions::default()).await;
        requester_conn.close(0, "done").await?;
        responder_conn.close(0, "done").await?;
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## Reply-to must be no-ack [#reply-to-must-be-no-ack]

Consuming `amq.rabbitmq.reply-to` requires `no-ack=true`. A manual-ack consume on the pseudo-queue is rejected with `406 precondition-failed` — a deliberate constraint of the direct-reply-to mechanism. The requester declares no reply queue at all; the connector mints the opaque `amq.rabbitmq.reply-to.g1.*` address per request, and the responder only ever sees that minted address, never the literal pseudo-queue name.

<Callout type="info">
  **No gRPC responder — RPC is fully in-protocol.** Unlike some connectors where the wire-protocol client cannot act as an RPC responder (forcing an embedded gRPC responder), AMQP RPC here is entirely in-protocol. No KubeMQ SDK, no gRPC, and no embedded responder are involved — the responder is a plain AMQP consumer on a KubeMQ Queue channel.
</Callout>

<Callout type="warn">
  **Direct reply-to is node-local in a cluster.** The `amq.rabbitmq.reply-to` pseudo-queue is single-node. In a cluster, the requester and responder must land on the **same** node (use load-balancer session affinity), or switch to an **explicit reply queue + correlation-id** instead. Single-node deployments are unaffected.
</Callout>

## Related [#related]

<Cards>
  <Card title="Work Queues" href="/connectors/rabbitmq/how-to/work-queues" description="Competing consumers on a durable queue — the request queue here is an ordinary work queue." />

  <Card title="Routing (direct)" href="/connectors/rabbitmq/how-to/routing" description="Selective delivery by exact routing key — the default exchange routes replies by queue name." />

  <Card title="Channel mapping" href="/connectors/rabbitmq/reference/channel-mapping" description="How the request queue and minted reply address resolve to KubeMQ Queue channels." />
</Cards>