KubeMQ
ConnectorsAWS (SQS & SNS)Concepts

Cross-Protocol Interop

Share an sqs.* channel between an AWS SDK app and a native KubeMQ gRPC/REST client — produce with boto3, consume with kubemq-go, and migrate one side at a time.

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

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.

DirectionProducerConsumerWhat carries over
AWS SDK → nativeSQS SendMessage on sqs.sharednative gRPC ReceiveQueueMessagesBody + sqs_* tags (sqs_message_id, sqs_sender_id, attribute tags)
Native → AWS SDKnative gRPC Send on sqs.sharedSQS ReceiveMessageBody + native tags; no sqs_* tags (see caveat)

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.

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 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 CreateQueued returns NonExistentQueue). Each client overrides only the endpoint (KUBEMQ_AWS_URL, default http://localhost:4566) and supplies dummy static credentials.

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,
		})
	}
}
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"])
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()));
            }
        }
    }
}
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);
});
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 });
}
# 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
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(())
}

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.

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")
}
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}")
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);
});

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.

Native-producer MessageId fallback

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 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.

Cluster caveat

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.

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.

SQS Queues

The point-to-point queue side that backs the shared sqs.* channel.

Channel mapping

The sqs.{name} grammar, the sqs_* tags, and the native-interop fallback caveat.

SNS Topics

The pub/sub side: virtual topics that fan out to sqs.* channels.

Was this page helpful?

Configuration

Why the KubeMQ AWS connector is opt-in, how CONNECTORS_AWS_ENABLE opens port 4566, and the accept-any vs static credential postures.

Getting Started

Enable the KubeMQ AWS connector, point a standard AWS SDK at port 4566, and run an SQS send-and-receive round-trip in minutes — no LocalStack, no KubeMQ SDK.

On this page

OverviewHow it worksThe AWS-SDK sideThe native consumerNative-producer MessageId fallbackCluster caveatNo RPC responderRelated