# Cross-Protocol Interop (/connectors/aws/concepts/cross-protocol-interop)



Because every SQS queue is a **normal KubeMQ Queue channel** (`sqs.{name}`), an AWS SDK application and a native KubeMQ gRPC/REST client can work the **same channel**. A message sent by `boto3` to `sqs.shared` is consumable by a native `kubemq-go` queue client on `sqs.shared` — and a message produced natively is receivable by an AWS SDK `ReceiveMessage`. This lets you **migrate one side at a time**, or run AWS-SDK producers alongside native KubeMQ consumers.

## Overview [#overview]

Both directions work on the same channel. The AWS-SDK side speaks the SQS HTTP protocol to the connector (port 4566); the native side speaks gRPC/REST to the KubeMQ broker directly. The channel — and therefore the message body — is shared.

| Direction        | Producer                           | Consumer                           | What carries over                                                       |
| ---------------- | ---------------------------------- | ---------------------------------- | ----------------------------------------------------------------------- |
| AWS SDK → native | SQS `SendMessage` on `sqs.shared`  | native gRPC `ReceiveQueueMessages` | Body + `sqs_*` tags (`sqs_message_id`, `sqs_sender_id`, attribute tags) |
| Native → AWS SDK | native gRPC `Send` on `sqs.shared` | SQS `ReceiveMessage`               | Body + native tags; **no** `sqs_*` tags (see caveat)                    |

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

The AWS connector registers `sqs.shared` and the native client connects to the same channel through the broker. A produce on either side is visible to a consume on the other.

<Mermaid
  chart="`
graph LR
SDK[&#x22;AWS SDK app<br/>(SendMessage / ReceiveMessage)&#x22;]
CONN[&#x22;AWS connector<br/>:4566&#x22;]
BROKER[&#x22;Message Broker&#x22;]
CH{{&#x22;Queue channel<br/>sqs.shared&#x22;}}
NATIVE[&#x22;Native KubeMQ client<br/>(gRPC / REST)&#x22;]

SDK -- &#x22;SQS HTTP&#x22; --> CONN
CONN -- &#x22;SendQueueMessage / Get&#x22; --> BROKER
BROKER --> CH
NATIVE -- &#x22;gRPC Send / Receive&#x22; --> BROKER
CH -. &#x22;shared messages&#x22; .-> NATIVE

class SDK,NATIVE client
class CONN connector
class BROKER broker
class CH queue
`"
/>

*The SQS queue `shared` and the native channel `sqs.shared` are the same KubeMQ Queue channel; an AWS-SDK producer and a native gRPC consumer share its messages — and vice versa.*

## The AWS-SDK side [#the-aws-sdk-side]

The AWS half is ordinary SQS code — `CreateQueue`, `SendMessage`, `ReceiveMessage`, `DeleteMessage` against `sqs.shared`. The queue must be **created via the AWS API first** (the registry is authoritative — a native channel never `CreateQueue`d returns `NonExistentQueue`). Each client overrides only the endpoint (`KUBEMQ_AWS_URL`, default `http://localhost:4566`) and supplies dummy static credentials.

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

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

    	"github.com/aws/aws-sdk-go-v2/aws"
    	"github.com/aws/aws-sdk-go-v2/config"
    	"github.com/aws/aws-sdk-go-v2/credentials"
    	"github.com/aws/aws-sdk-go-v2/service/sqs"
    )

    func main() {
    	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
    	defer cancel()
    	url := os.Getenv("KUBEMQ_AWS_URL")
    	if url == "" {
    		url = "http://localhost:4566"
    	}
    	cfg, _ := config.LoadDefaultConfig(ctx,
    		config.WithRegion("us-east-1"),
    		config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")),
    	)
    	sdk := sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(url) })

    	// The channel sqs.shared must exist in the registry — create it via the API.
    	created, err := sdk.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String("shared")})
    	if err != nil {
    		log.Fatalf("CreateQueue: %v", err)
    	}
    	queueURL := aws.ToString(created.QueueUrl)
    	fmt.Printf("CreateQueue: %s (native channel: sqs.shared)\n", queueURL)

    	// Direction 1: SDK produces; a native gRPC client consumes on sqs.shared.
    	if _, err := sdk.SendMessage(ctx, &sqs.SendMessageInput{
    		QueueUrl:    aws.String(queueURL),
    		MessageBody: aws.String("from the AWS SDK"),
    	}); err != nil {
    		log.Fatalf("SendMessage: %v", err)
    	}
    	fmt.Println("[SDK -> native] sent \"from the AWS SDK\" to sqs.shared")

    	// Direction 2: the AWS SDK receives a message a native client produced.
    	recv, err := sdk.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
    		QueueUrl:        aws.String(queueURL),
    		WaitTimeSeconds: 10,
    	})
    	if err != nil {
    		log.Fatalf("ReceiveMessage: %v", err)
    	}
    	if len(recv.Messages) == 1 {
    		m := recv.Messages[0]
    		fmt.Printf("[native -> SDK] received %q MessageId=%s\n", aws.ToString(m.Body), aws.ToString(m.MessageId))
    		_, _ = sdk.DeleteMessage(ctx, &sqs.DeleteMessageInput{
    			QueueUrl: aws.String(queueURL), ReceiptHandle: m.ReceiptHandle,
    		})
    	}
    }
    ```
  </Tab>

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

    import boto3

    sqs = boto3.client(
        "sqs",
        endpoint_url=os.environ.get("KUBEMQ_AWS_URL", "http://localhost:4566"),
        region_name="us-east-1",
        aws_access_key_id="test",
        aws_secret_access_key="test",
    )

    # The channel sqs.shared must exist in the registry — create it via the API.
    url = sqs.create_queue(QueueName="shared")["QueueUrl"]
    print(f"CreateQueue -> {url} (native channel: sqs.shared)")

    # Direction 1: SDK produces; a native gRPC client consumes on sqs.shared.
    sqs.send_message(QueueUrl=url, MessageBody="from-aws-sdk")
    print("[SDK -> native] sent 'from-aws-sdk' to sqs.shared")

    # Direction 2: the AWS SDK receives a message a native client produced.
    recv = sqs.receive_message(QueueUrl=url, MaxNumberOfMessages=1, WaitTimeSeconds=10)
    for m in recv.get("Messages", []):
        print(f"[native -> SDK] received {m['Body']!r} MessageId={m['MessageId']}")
        sqs.delete_message(QueueUrl=url, ReceiptHandle=m["ReceiptHandle"])
    ```
  </Tab>

  <Tab value="Java">
    ```java
    import java.net.URI;

    import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
    import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
    import software.amazon.awssdk.regions.Region;
    import software.amazon.awssdk.services.sqs.SqsClient;
    import software.amazon.awssdk.services.sqs.model.Message;

    public final class Main {
        public static void main(String[] args) {
            String url = System.getenv().getOrDefault("KUBEMQ_AWS_URL", "http://localhost:4566");
            try (SqsClient sqs = SqsClient.builder()
                    .endpointOverride(URI.create(url)).region(Region.US_EAST_1)
                    .credentialsProvider(StaticCredentialsProvider.create(
                            AwsBasicCredentials.create("test", "test")))
                    .build()) {

                // The channel sqs.shared must exist in the registry — create it via the API.
                String queueUrl = sqs.createQueue(b -> b.queueName("shared")).queueUrl();
                System.out.println("CreateQueue -> " + queueUrl + " (native channel: sqs.shared)");

                // Direction 1: SDK produces; a native gRPC client consumes on sqs.shared.
                sqs.sendMessage(b -> b.queueUrl(queueUrl).messageBody("from the AWS SDK"));
                System.out.println("[SDK -> native] sent to sqs.shared");

                // Direction 2: the AWS SDK receives a message a native client produced.
                var recv = sqs.receiveMessage(b -> b.queueUrl(queueUrl).waitTimeSeconds(10));
                for (Message m : recv.messages()) {
                    System.out.println("[native -> SDK] received '" + m.body()
                            + "' MessageId=" + m.messageId());
                    sqs.deleteMessage(b -> b.queueUrl(queueUrl).receiptHandle(m.receiptHandle()));
                }
            }
        }
    }
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import {
      SQSClient,
      CreateQueueCommand,
      SendMessageCommand,
      ReceiveMessageCommand,
      DeleteMessageCommand,
    } from "@aws-sdk/client-sqs";

    const sqs = new SQSClient({
      endpoint: process.env["KUBEMQ_AWS_URL"] ?? "http://localhost:4566",
      region: "us-east-1",
      credentials: { accessKeyId: "test", secretAccessKey: "test" },
    });

    async function main(): Promise<void> {
      // The channel sqs.shared must exist in the registry — create it via the API.
      const queueUrl = (await sqs.send(new CreateQueueCommand({ QueueName: "shared" }))).QueueUrl!;
      console.log(`CreateQueue -> ${queueUrl} (native channel: sqs.shared)`);

      // Direction 1: SDK produces; a native gRPC client consumes on sqs.shared.
      await sqs.send(new SendMessageCommand({ QueueUrl: queueUrl, MessageBody: "order from AWS SDK" }));
      console.log("[SDK -> native] sent to sqs.shared");

      // Direction 2: the AWS SDK receives a message a native client produced.
      const recv = await sqs.send(new ReceiveMessageCommand({ QueueUrl: queueUrl, WaitTimeSeconds: 10 }));
      for (const m of recv.Messages ?? []) {
        console.log(`[native -> SDK] received "${m.Body}" MessageId=${m.MessageId}`);
        await sqs.send(new DeleteMessageCommand({ QueueUrl: queueUrl, ReceiptHandle: m.ReceiptHandle! }));
      }
    }

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

  <Tab value="C#">
    ```csharp
    using Amazon.Runtime;
    using Amazon.SQS;
    using Amazon.SQS.Model;

    var url = Environment.GetEnvironmentVariable("KUBEMQ_AWS_URL") ?? "http://localhost:4566";
    using var sqs = new AmazonSQSClient(new BasicAWSCredentials("test", "test"),
        new AmazonSQSConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" });

    // The channel sqs.shared must exist in the registry — create it via the API.
    var queueUrl = (await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = "shared" })).QueueUrl;
    Console.WriteLine($"CreateQueue -> {queueUrl} (native channel: sqs.shared)");

    // Direction 1: SDK produces; a native gRPC client consumes on sqs.shared.
    await sqs.SendMessageAsync(new SendMessageRequest { QueueUrl = queueUrl, MessageBody = "from the AWS SDK" });
    Console.WriteLine("[SDK -> native] sent to sqs.shared");

    // Direction 2: the AWS SDK receives a message a native client produced.
    var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest { QueueUrl = queueUrl, WaitTimeSeconds = 10 });
    foreach (var m in recv.Messages)
    {
        Console.WriteLine($"[native -> SDK] received '{m.Body}' MessageId={m.MessageId}");
        await sqs.DeleteMessageAsync(new DeleteMessageRequest { QueueUrl = queueUrl, ReceiptHandle = m.ReceiptHandle });
    }
    ```
  </Tab>

  <Tab value="Ruby">
    ```ruby
    # frozen_string_literal: true
    require "aws-sdk-sqs"

    Aws::SQS::Client.remove_plugin(Aws::SQS::Plugins::QueueUrls)

    url = ENV.fetch("KUBEMQ_AWS_URL", "http://localhost:4566")
    sqs = Aws::SQS::Client.new(
      endpoint: url, region: "us-east-1", access_key_id: "test", secret_access_key: "test"
    )

    # The channel sqs.shared must exist in the registry — create it via the API.
    queue_url = sqs.create_queue(queue_name: "shared").queue_url
    puts "CreateQueue -> #{queue_url} (native channel: sqs.shared)"

    # Direction 1: SDK produces; a native gRPC client consumes on sqs.shared.
    sqs.send_message(queue_url: queue_url, message_body: "from the AWS SDK")
    puts "[SDK -> native] sent to sqs.shared"

    # Direction 2: the AWS SDK receives a message a native client produced.
    recv = sqs.receive_message(queue_url: queue_url, max_number_of_messages: 1, wait_time_seconds: 10)
    recv.messages.each do |m|
      puts "[native -> SDK] received #{m.body.inspect} MessageId=#{m.message_id}"
      sqs.delete_message(queue_url: queue_url, receipt_handle: m.receipt_handle)
    end
    ```
  </Tab>

  <Tab value="Rust">
    ```rust
    use aws_config::{BehaviorVersion, Region};
    use aws_credential_types::Credentials;
    use std::error::Error;

    #[tokio::main]
    async fn main() -> Result<(), Box<dyn Error>> {
        let url = std::env::var("KUBEMQ_AWS_URL").unwrap_or_else(|_| "http://localhost:4566".into());
        let conf = aws_config::defaults(BehaviorVersion::latest())
            .region(Region::new("us-east-1"))
            .credentials_provider(Credentials::new("test", "test", None, None, "static"))
            .endpoint_url(url)
            .load()
            .await;
        let sqs = aws_sdk_sqs::Client::new(&conf);

        // The channel sqs.shared must exist in the registry — create it via the API.
        let queue_url = sqs.create_queue().queue_name("shared").send().await?.queue_url.unwrap();
        println!("CreateQueue -> {queue_url} (native channel: sqs.shared)");

        // Direction 1: SDK produces; a native gRPC client consumes on sqs.shared.
        sqs.send_message().queue_url(&queue_url).message_body("from the AWS SDK").send().await?;
        println!("[SDK -> native] sent to sqs.shared");

        // Direction 2: the AWS SDK receives a message a native client produced.
        let recv = sqs.receive_message().queue_url(&queue_url).wait_time_seconds(10).send().await?;
        for m in recv.messages() {
            println!("[native -> SDK] received '{}' MessageId={}",
                m.body().unwrap_or_default(), m.message_id().unwrap_or("<none>"));
            if let Some(handle) = m.receipt_handle() {
                sqs.delete_message().queue_url(&queue_url).receipt_handle(handle).send().await?;
            }
        }
        Ok(())
    }
    ```
  </Tab>
</Tabs>

## The native consumer [#the-native-consumer]

The other side of the channel is an ordinary KubeMQ **Queue** client talking gRPC to the broker (default `localhost:50000`) on the `sqs.shared` channel — no AWS SDK involved. A message the AWS SDK sent carries the connector's `sqs_*` tags (`sqs_message_id`, attribute tags); a message the native client sends is later receivable by an SQS `ReceiveMessage`.

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

    import (
    	"context"
    	"fmt"
    	"log"
    	"time"

    	kubemq "github.com/kubemq-io/kubemq-go"
    )

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

    	// Native KubeMQ gRPC queue client on the shared channel sqs.shared.
    	native, err := kubemq.NewQueuesClient(ctx,
    		kubemq.WithAddress("localhost", 50000),
    		kubemq.WithClientId("kubemq-aws-interop-native"),
    		kubemq.WithTransportType(kubemq.TransportTypeGRPC),
    	)
    	if err != nil {
    		log.Fatalf("connect native gRPC: %v", err)
    	}
    	defer func() { _ = native.Close() }()

    	// Consume a message the AWS SDK produced on sqs.shared.
    	pull, err := native.Pull(ctx, kubemq.NewReceiveQueueMessagesRequest().
    		SetClientId("kubemq-aws-interop-native").
    		SetChannel("sqs.shared").
    		SetMaxNumberOfMessages(1).
    		SetWaitTimeSeconds(10))
    	if err != nil || pull.IsError {
    		log.Fatalf("native Pull: %v %s", err, pull.Error)
    	}
    	for _, m := range pull.Messages {
    		fmt.Printf("native received %q (sqs_message_id=%s)\n", string(m.Body), m.Tags["sqs_message_id"])
    	}

    	// Produce a message the AWS SDK can ReceiveMessage on sqs.shared.
    	if _, err := native.Send(ctx, kubemq.NewQueueMessage().
    		SetChannel("sqs.shared").
    		SetBody([]byte("from native gRPC"))); err != nil {
    		log.Fatalf("native Send: %v", err)
    	}
    	fmt.Println("native sent \"from native gRPC\" to sqs.shared")
    }
    ```
  </Tab>

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

    from kubemq import QueueMessage, QueuesClient

    CHANNEL = "sqs.shared"
    GRPC_ADDRESS = os.environ.get("KUBEMQ_GRPC_ADDRESS", "localhost:50000")

    # Native KubeMQ gRPC queue client on the shared channel sqs.shared.
    with QueuesClient(address=GRPC_ADDRESS, client_id="kubemq-aws-interop-python") as native:
        # Consume a message the AWS SDK produced on sqs.shared.
        resp = native.receive_queue_messages(channel=CHANNEL, max_messages=1, wait_timeout_in_seconds=10)
        for msg in resp.messages:
            body = msg.body.decode("utf-8")
            print(f"native received {body!r} sqs_message_id={msg.tags.get('sqs_message_id')}")
            msg.ack()

        # Produce a message the AWS SDK can ReceiveMessage on sqs.shared.
        result = native.send_queue_message(
            QueueMessage(channel=CHANNEL, body=b"from-native-grpc", tags={"origin": "native"})
        )
        print(f"native sent -> id={result.id}")
    ```
  </Tab>

  <Tab value="JavaScript">
    ```typescript
    import { KubeMQClient, createQueueMessage, bytesToString } from "kubemq-js";

    const CHANNEL = "sqs.shared";
    const address = process.env["KUBEMQ_BROKER_ADDRESS"] ?? "localhost:50000";

    async function main(): Promise<void> {
      // Native KubeMQ gRPC queue client on the shared channel sqs.shared.
      const native = await KubeMQClient.create({ address, clientId: "kubemq-aws-interop-js" });
      try {
        // Consume a message the AWS SDK produced on sqs.shared.
        const msgs = await native.receiveQueueMessages({ channel: CHANNEL, maxMessages: 1, waitTimeoutSeconds: 10 });
        for (const m of msgs) {
          console.log(`native received "${bytesToString(m.body)}" sqs_message_id=${m.tags["sqs_message_id"] ?? "<none>"}`);
        }

        // Produce a message the AWS SDK can ReceiveMessage on sqs.shared.
        await native.sendQueueMessage(createQueueMessage({ channel: CHANNEL, body: "from native gRPC" }));
        console.log("native sent to sqs.shared");
      } finally {
        await native.close();
      }
    }

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

<Callout type="info">
  The native consumer is the **only** place a KubeMQ SDK appears in the AWS connector examples — the AWS-SDK half above is idiomatic AWS code in every language. The native half ships in all seven languages; where a language's native queue client is less mature, it can fall back to a `kubemq-go` sidecar or a REST queue call.
</Callout>

## Native-producer MessageId fallback [#native-producer-messageid-fallback]

<Callout type="warn">
  A message produced by a **native** KubeMQ client on `sqs.*` lacks the connector's `sqs_*` tags. On the SQS receive side this means:

  * its `MessageId` **falls back to the broker message id** (not a connector-minted UUID);
  * it has &#x2A;*no `SenderId`** (there is no authenticated `sqs_sender_id`);
  * **no policy stamping** is applied (no `RedrivePolicy` `MaxReceiveCount` / source-queue tags).

  This is expected and harmless for interop — the body and tags round-trip — but do not assume an SQS-style `MessageId` / `SenderId` on messages that entered the channel natively.
</Callout>

## Cluster caveat [#cluster-caveat]

<Callout type="warn">
  **Receipt handles are node-local.** When an AWS SDK consumer and a native client share a channel across cluster nodes, the AWS consumer's receipt handle is only valid on the node that issued it. Use a **sticky load balancer** (session affinity) so a consumer's receive, delete, and visibility-change calls all land on the same node. See the [connectivity and security guide](/connectors/aws/how-to/connectivity-and-security).
</Callout>

## No RPC responder [#no-rpc-responder]

SQS and SNS are queue / pub-sub, **not** request/reply — there is **no gRPC RPC responder** anywhere in the AWS connector. Cross-protocol interop is a native **queue** client (produce / consume), not an RPC responder.

## Related [#related]

<Cards>
  <Card title="SQS Queues" href="/connectors/aws/how-to/sqs-queues" description="The point-to-point queue side that backs the shared sqs.* channel." />

  <Card title="Channel mapping" href="/connectors/aws/reference/channel-mapping" description="The sqs.{name} grammar, the sqs_* tags, and the native-interop fallback caveat." />

  <Card title="SNS Topics" href="/connectors/aws/how-to/sns-topics" description="The pub/sub side: virtual topics that fan out to sqs.* channels." />
</Cards>