# Migrating from AWS SQS/SNS (/connectors/aws/reference/migration-from-aws)
**Point your existing AWS SQS/SNS application at KubeMQ by changing only the endpoint URL.**
Same AWS SDK, same code, same SQS and SNS wire protocols. There is no SDK to adopt, no proto,
no data migration — SQS data lives in normal KubeMQ Queue channels. Rollback is **config-only**
(`CONNECTORS_AWS_ENABLE=false`).
If you already run an SDK against a **LocalStack** endpoint, the switch is the same single
variable — point it at the KubeMQ AWS connector instead of LocalStack.
But several connector behaviors **deviate from real AWS**. Read the deviations below before you
migrate; most are invisible until a corner case hits production.
## Overview [#overview]
The AWS connector exposes the real AWS SQS and SNS wire protocols (the AWS JSON and Query
protocols) over HTTP, so unmodified AWS SDK clients — `boto3`, the AWS SDK for Go/JavaScript,
the AWS CLI — talk to KubeMQ without any code changes. An endpoint-override environment
variable is all that changes on the client side.
SQS queues map onto native KubeMQ Queue channels (`sqs.{name}`), making AWS producers and
native gRPC/REST consumers interoperable on the same messages. SNS topics are virtual
(registry-only) and fan out to SQS subscriptions and HTTP/HTTPS webhook endpoints. Requests are
authenticated with AWS Signature V4.
* **Canonical client:** AWS SDK `boto3` 1.x (Python).
* **Port:** `4566` (HTTP).
* **Drop-in level:** endpoint-only — change the SDK endpoint override; no application code
changes.
Port `4566` is a **client-side endpoint convention** (the LocalStack / SDK default), not a fixed
KubeMQ listener — the actual connector port is `Connectors.Aws.Port` and can be changed. That is
why it does not appear in KubeMQ's shared broker-ports tables, which track fixed listeners only.
The connector is **opt-in — disabled by default** (`Connectors.Aws.Enable = false`). Enable it
with its enable variable; this opens the HTTP listener on port `4566`:
## Compatibility Matrix [#compatibility-matrix]
The column below is the AWS SQS/SNS slice of the
[master cross-protocol matrix](/connectors/how-to/migration).
| Dimension | AWS SQS/SNS |
| -------------------------------------- | --------------------------------------------------------------------- |
| **Drop-in level** | endpoint-only |
| **Point-to-point queues** | ✅ SQS |
| **Pub/sub (non-durable)** | ✅ SNS |
| **Durable / persistent subscriptions** | ✅ (SQS durable) |
| **Request/reply (RPC)** | N/A (no RPC) |
| **Ordering guarantee** | ✅ FIFO |
| **Transactions** | N/A |
| **Dead-letter / redrive** | ✅ redrive + move-task |
| **Selectors / filtering / wildcards** | ✅ SNS filter policies¹ |
| **Auth model** | SigV4 / accept-any |
| **TLS / mTLS** | ❌ connector HTTP-only² |
| **Top unsupported** | SNS email/SMS/Lambda/push; queue/topic IAM policies; TLS at connector |
¹ SNS filter policies work with the `MessageAttributes` scope only; the `MessageBody`
scope is rejected (`InvalidParameter`).
² The connector listens on plain HTTP. Terminate TLS at a reverse proxy (see
[Security](#tls--terminate-at-a-reverse-proxy)).
## Connection / Endpoint Migration [#connection--endpoint-migration]
No code changes are required. Set the AWS SDK endpoint-override environment variables to point
at the KubeMQ host:
```bash title="Terminal"
# Before (real AWS): no override; the SDK uses the regional AWS endpoint.
# After (KubeMQ):
export AWS_ENDPOINT_URL_SQS=http://kubemq-host:4566
export AWS_ENDPOINT_URL_SNS=http://kubemq-host:4566
export AWS_ACCESS_KEY_ID=AKIAEXAMPLE
export AWS_SECRET_ACCESS_KEY=secret
export AWS_DEFAULT_REGION=kubemq # any region value works; the region segment is not enforced
```
All current AWS SDKs and the AWS CLI honor `AWS_ENDPOINT_URL_SQS` / `AWS_ENDPOINT_URL_SNS`. If
your SDK version predates these variables, use the per-client endpoint override instead:
```python
import boto3
sqs = boto3.client(
"sqs",
endpoint_url="http://kubemq-host:4566",
region_name="kubemq",
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="secret",
)
```
In the default **accept-any** mode the connector does not cryptographically verify the
signature — but the SDK must still form a **syntactically valid** SigV4 request whose
credential-scope service is `sqs` or `sns`. So you must give the SDK a dummy access key, secret,
and region (any values); omitting them yields a "missing credentials" SDK error. The only
SigV4-exempt action is SNS `ConfirmSubscription`.
### Credential mapping [#credential-mapping]
For each `AccessKeyId` your application uses, add a credential entry to the KubeMQ
configuration. The `ClientID` field maps that key to a KubeMQ identity for authorization and
audit:
```toml title="config.toml"
[Connectors.Aws]
Enable = true
Port = "4566"
[[Connectors.Aws.Credentials]]
AccessKeyId = "AKIAEXAMPLE"
SecretAccessKey = "secret"
ClientID = "billing-service" # optional; defaults to AccessKeyId
```
### Recreate resources [#recreate-resources]
Queues, topics, subscriptions, and their attributes/tags must be recreated through the AWS API
against KubeMQ. The registry is authoritative — existing SQS/SNS resources from real AWS are
**not** migrated automatically.
## Concept & Destination Mapping [#concept--destination-mapping]
| AWS concept | KubeMQ pattern | KubeMQ channel |
| ---------------------- | ----------------------- | ------------------------------------------------- |
| SQS standard queue | Queues | `sqs.{queue-name}` |
| SQS FIFO queue | Queues (per group) | `sqs.{queue-name}` per `MessageGroupId` |
| SNS topic | Virtual (registry-only) | `sns.{topic-name}` (authorization pseudo-channel) |
| SNS subscription → SQS | Queues (fan-out) | `sqs.{target-queue-name}` |
**FIFO ordering:** each `MessageGroupId` maps to its own KubeMQ Queue channel, preserving
per-group ordering. `MessageGroupId` is required on every FIFO send.
**Message attributes** round-trip losslessly through KubeMQ message Tags
(`sqs_attr_{Name}` = `{DataType}|{value}`). The connector also stamps `sqs_message_id`,
`sqs_sender_id` (the authenticated `ClientID`), and `sqs_trace_header` when present.
**Native interop:** native KubeMQ gRPC/REST clients can produce and consume on `sqs.*` channels
directly. Natively produced messages lack `sqs_*` tags; `MessageId` falls back to the broker
MessageID and no per-message redrive policy is stamped (see [deviations](#native-producers-bypass-sqs-policy-stamping)).
**ARNs** use the configured `Region` (default `kubemq`) and `AccountId` (default
`000000000000`). Set `Connectors.Aws.Region` / `Connectors.Aws.AccountId` if your tooling
validates ARN format.
## Canonical Client Example [#canonical-client-example]
The examples use the AWS SDK `boto3` 1.x: `boto3.client`, `create_queue`, `send_message`,
`receive_message`, `delete_message`, `create_topic`, `subscribe`, and `publish`.
### SQS — create, send, receive, delete [#sqs--create-send-receive-delete]
```python
import boto3
# boto3 1.x — point the SDK at KubeMQ
sqs = boto3.client(
"sqs",
endpoint_url="http://kubemq-host:4566",
region_name="kubemq",
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="secret",
)
# Create a standard queue (idempotent — returns the URL if it already exists)
resp = sqs.create_queue(QueueName="orders")
queue_url = resp["QueueUrl"]
# Send a message
sqs.send_message(
QueueUrl=queue_url,
MessageBody='{"orderId": "A-001", "amount": 99.95}',
MessageAttributes={
"source": {"DataType": "String", "StringValue": "checkout-service"},
},
)
# Receive (long-poll up to 20 s)
resp = sqs.receive_message(
QueueUrl=queue_url,
MaxNumberOfMessages=1,
WaitTimeSeconds=20,
MessageAttributeNames=["All"],
)
for msg in resp.get("Messages", []):
print(f"Received: {msg['Body']}")
# Acknowledge by deleting
sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msg["ReceiptHandle"])
```
### FIFO queue [#fifo-queue]
```python
# Create a FIFO queue (name must end in .fifo)
resp = sqs.create_queue(
QueueName="orders.fifo",
Attributes={
"FifoQueue": "true",
"ContentBasedDeduplication": "true",
},
)
fifo_url = resp["QueueUrl"]
# Send to a specific message group (preserves ordering per group)
sqs.send_message(
QueueUrl=fifo_url,
MessageBody='{"orderId": "A-002"}',
MessageGroupId="region-us-east",
)
```
### Dead-letter queue (redrive) [#dead-letter-queue-redrive]
```python
import json
# 1. Create the DLQ
dlq_resp = sqs.create_queue(QueueName="orders-dlq")
dlq_url = dlq_resp["QueueUrl"]
dlq_arn = sqs.get_queue_attributes(
QueueUrl=dlq_url, AttributeNames=["QueueArn"]
)["Attributes"]["QueueArn"]
# 2. Attach a redrive policy to the source queue
sqs.set_queue_attributes(
QueueUrl=queue_url,
Attributes={
"RedrivePolicy": json.dumps({
"deadLetterTargetArn": dlq_arn,
"maxReceiveCount": "3",
}),
},
)
# After 3 failed receives the broker automatically moves the message to orders-dlq.
```
### SNS fan-out (topic → SQS) [#sns-fan-out-topic--sqs]
```python
sns = boto3.client(
"sns",
endpoint_url="http://kubemq-host:4566",
region_name="kubemq",
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="secret",
)
# Create the topic and subscribe the SQS queue
topic_arn = sns.create_topic(Name="product-events")["TopicArn"]
orders_arn = sqs.get_queue_attributes(
QueueUrl=queue_url, AttributeNames=["QueueArn"]
)["Attributes"]["QueueArn"]
sns.subscribe(
TopicArn=topic_arn,
Protocol="sqs", # only 'sqs' and 'http'/'https' are supported
Endpoint=orders_arn,
)
# Publish — the message fans out to all confirmed subscriptions
sns.publish(
TopicArn=topic_arn,
Message='{"event": "product.created", "id": "P-100"}',
MessageAttributes={
"category": {"DataType": "String", "StringValue": "electronics"},
},
)
```
## Security [#security]
### Authentication — SigV4 [#authentication--sigv4]
The connector verifies AWS Signature V4 (`Authorization: AWS4-HMAC-SHA256 …`). Header-style and
query-string-style signatures are both supported, with a ±15-minute clock-skew window.
* Credentials are configured under `Connectors.Aws.Credentials` (or `CredentialsData` for
environment / Kubernetes Secret injection).
* **Accept-any mode:** if no credentials are configured the connector only parses the
`AccessKeyId` and uses it as the `ClientID`. This is intended for local development only and is
logged once at startup.
* Failure cases: unknown key → HTTP 403 `InvalidClientTokenId`; bad signature / clock skew →
HTTP 403 `SignatureDoesNotMatch`; malformed header → HTTP 400 `IncompleteSignature`.
* `ConfirmSubscription` requests with no `Authorization` header bypass SigV4 — the single-purpose
confirmation token is the authenticator (AWS parity). No other action is exempt.
### TLS — terminate at a reverse proxy [#tls--terminate-at-a-reverse-proxy]
The connector listens on plain **HTTP only**. It does **not** support TLS natively. To secure
traffic in transit, place a TLS-terminating reverse proxy (nginx, Envoy, HAProxy, an AWS ALB, …)
in front of the connector port, and configure clients to target the proxy's HTTPS endpoint.
### Optional SNS message signing (off by default) [#optional-sns-message-signing-off-by-default]
By default (`Connectors.Aws.MessageSigning = false`) SNS notification envelopes are unsigned.
Set `Connectors.Aws.MessageSigning = true` to emit SigV2 RSA-SHA256 signatures. Note that the
signing certificate is self-signed and not Amazon-rooted; SDK verifiers that pin Amazon's cert
chain will still reject the signature. See also
[SNS notification signatures are unsigned by default](#sns-notification-signatures-are-unsigned-by-default).
### Authorization (Casbin) [#authorization-casbin]
When the KubeMQ authorization service is enabled, every channel-mapped operation is enforced
against the existing Casbin policy:
| Operation class | Casbin check |
| ------------------------------------------------------------------------- | ------------------------ |
| `SendMessage`, `SendMessageBatch`, SNS `Publish` (per matched SQS target) | `write` on `sqs.{queue}` |
| `ReceiveMessage`, `DeleteMessage`, `ChangeMessageVisibility` | `read` on `sqs.{queue}` |
| Queue management (Create / Delete / Purge / Set / Tag) | `write` on `sqs.{queue}` |
| Topic & subscription management | `write` on `sns.{topic}` |
`ListQueues`, `ListTopics`, `ListSubscriptions`, and `GetQueueUrl` are allowed for any
authenticated principal and return unfiltered results. See
[Authentication & security](/connectors/reference/auth-and-security) for policy configuration.
## What Does NOT Migrate / Deviations [#what-does-not-migrate--deviations]
### Unsupported SNS subscription protocols [#unsupported-sns-subscription-protocols]
Only `sqs` and `http`/`https` subscription protocols are supported. The following AWS SNS
delivery targets are **not supported** and return `InvalidParameter`:
* Email / email-JSON
* SMS
* AWS Lambda
* Mobile push (APNs, GCM, ADM, Baidu)
FIFO topics additionally restrict subscriptions to `sqs` only (`http`/`https` are rejected for
`.fifo` topics).
### No TLS at the connector [#no-tls-at-the-connector]
The connector is HTTP-only (see [TLS — terminate at a reverse proxy](#tls--terminate-at-a-reverse-proxy)).
In particular, do not configure clients to send to `https://kubemq-host:4566` directly.
### Queue/topic IAM policies ignored [#queuetopic-iam-policies-ignored]
`Policy` fields on `SetQueueAttributes` and `SetTopicAttributes` are accepted but **not
enforced**. Authorization is handled by KubeMQ's Casbin engine (see above).
### FIFO `SequenceNumber` is broker-derived [#fifo-sequencenumber-is-broker-derived]
FIFO `SequenceNumber` is derived from the broker-assigned timestamp and zero-padded to 20
digits. On send it reflects the broker send-timestamp; on receive it reflects the true broker
sequence (still per-group increasing). It is **not** a monotonic counter matching AWS semantics —
do not use it for ordering comparisons across producers.
### FIFO topic `ContentBasedDeduplication` not supported [#fifo-topic-contentbaseddeduplication-not-supported]
Topic-level `ContentBasedDeduplication` on a FIFO topic is unsupported: setting it returns
`InvalidParameter` and getting it always reads `"false"`. Deduplication is enforced on the target
FIFO queues, not at the topic level — pass an explicit `MessageDeduplicationId` instead.
### Message retention clamped [#message-retention-clamped]
`MessageRetentionPeriod` is clamped to `Connectors.Aws.MaxExpirationSeconds` at send time.
Changes to `SetQueueAttributes` retention are **not** retroactive to messages already in the
queue (AWS applies them retroactively).
### SNS notification signatures are unsigned by default [#sns-notification-signatures-are-unsigned-by-default]
By default (`Connectors.Aws.MessageSigning = false`) the SNS notification envelope is unsigned
(`SignatureVersion: "1"`, empty `Signature` / `SigningCertURL`) — webhook consumers must skip
verification. Set `Connectors.Aws.MessageSigning = true` to emit `SignatureVersion: "2"` SigV2
RSA-SHA256 signatures (self-signed cert; not Amazon-rooted, so SDK verifiers that pin Amazon's
cert chain still won't validate it).
### Receipt handles and the in-flight tracker are node-local [#receipt-handles-and-the-in-flight-tracker-are-node-local]
In a cluster, `DeleteMessage` and `ChangeMessageVisibility` must reach the **same node** that
served the `ReceiveMessage`. The receipt handle encodes the node ID; other nodes return
`ReceiptHandleIsInvalid`. Use session-sticky (source-IP or connection-sticky) load balancing, or
pin each consumer to one node.
### Hard crash loses in-flight messages and pending webhook retries [#hard-crash-loses-in-flight-messages-and-pending-webhook-retries]
KubeMQ's downstream read is destructive. Only the graceful-shutdown path returns in-flight
messages to their queues; SNS delivery/retry state is in-memory on the publishing node, so a
hard kill loses it.
### Native producers bypass SQS policy stamping [#native-producers-bypass-sqs-policy-stamping]
Messages published to `sqs.*` channels by native KubeMQ gRPC/REST clients carry no per-message
retention or redrive policy and no `sqs_*` identity tags.
### Empty-queue short-poll latency [#empty-queue-short-poll-latency]
A `ReceiveMessage` with `WaitTimeSeconds=0` on an empty queue returns within \~1 second instead of
immediately (the broker's downstream wait granularity has a 1-second minimum).
### Other out-of-scope operations [#other-out-of-scope-operations]
The following simply **won't work** (see
[Capabilities](/connectors/aws/reference/capabilities)): KMS / SSE,
`AddPermission` / `RemovePermission`, SQS message-move tasks, signed-notification verification,
extended-client messages over 256 KiB, and CloudWatch metrics emulation. Cross-account is
unsupported — `QueueOwnerAWSAccountId` is accepted and ignored; there is a single configurable
`AccountId`.
## Verification Smoke Test [#verification-smoke-test]
Run this after enabling the connector (`CONNECTORS_AWS_ENABLE=true`) and recreating any needed
queues. It is the same `boto3` 1.x code as a one-pass send → receive → delete confirmation:
```python
import boto3
ENDPOINT = "http://kubemq-host:4566"
CREDS = dict(
endpoint_url=ENDPOINT,
region_name="kubemq",
aws_access_key_id="AKIAEXAMPLE",
aws_secret_access_key="secret",
)
sqs = boto3.client("sqs", **CREDS)
# 1. Ensure the queue exists
queue_url = sqs.create_queue(QueueName="smoke-test")["QueueUrl"]
# 2. Publish one message
sqs.send_message(QueueUrl=queue_url, MessageBody="smoke-test-payload")
print("Sent: smoke-test-payload")
# 3. Receive and confirm arrival
resp = sqs.receive_message(QueueUrl=queue_url, WaitTimeSeconds=5)
msgs = resp.get("Messages", [])
assert msgs, "ERROR: no message received"
assert msgs[0]["Body"] == "smoke-test-payload", f"Unexpected body: {msgs[0]['Body']}"
print(f"Received: {msgs[0]['Body']}")
# 4. Acknowledge
sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=msgs[0]["ReceiptHandle"])
print("Acknowledged. Smoke test PASSED.")
```
## See Also [#see-also]