# Fan-Out (SNS → SQS) (/connectors/aws/how-to/fan-out) **Fan-out** delivers one published message to many subscribers. An SNS `Publish` resolves, at publish time, to every **confirmed**, filter-matching subscription and delivers a copy to each — SQS queues (in a single batch send) and HTTP/HTTPS webhooks. This is the classic "one event, many consumers" pattern: an order-placed event reaches the billing queue, the shipping queue, and an analytics webhook in a single publish. ## Overview [#overview] `CreateQueue` each target queue, `CreateTopic`, then `Subscribe(Protocol=sqs, Endpoint=)` for each — an `sqs` subscription **auto-confirms** immediately. `Publish` then fans out to every matching subscription. | Step | Action | Behavior | | ---------------- | ------------------------------- | --------------------------------------------------------- | | Targets | `CreateQueue` ×N | Each becomes channel `sqs.{name}` | | Topic | `CreateTopic` | Virtual registry entry | | Subscribe | `Subscribe(Protocol=sqs)` | Auto-confirmed; `http`/`https` need `ConfirmSubscription` | | Publish | `Publish` / `PublishBatch` | One `MessageId`, shared across all deliveries | | SQS delivery | Single `SendQueueMessagesBatch` | All `sqs` targets of that publish in one batch | | Webhook delivery | In-memory delivery engine | Retry → circuit breaker → DLQ | Fan-out semantics: * **One `MessageId` per publish**, shared across all deliveries. * **Zero matching subscriptions → the publish still succeeds** and the message is dropped (no error). * **Per-target failures** (deleted queue, unauthorized, oversize, FIFO mismatch) are dropped with a metric and do **not** fail the publish; a per-target authorization check applies on each `sqs.{queue}`. ## How it works [#how-it-works] A single publish resolves to the set of confirmed, filter-matching subscriptions; all SQS targets go out in one batch send and webhooks go through the delivery engine. Every delivery shares the same `MessageId`. *One publish fans out to every confirmed subscription — SQS queues in a single batch send and HTTP/HTTPS webhooks through the delivery engine — all sharing one `MessageId`; a filtered subscription receives the copy only when its `FilterPolicy` matches.* ## Fan one publish to many queues [#fan-one-publish-to-many-queues] Subscribe two SQS queues to a topic, publish once, and watch both queues receive the same `MessageId`. Each client overrides only the endpoint (`KUBEMQ_AWS_URL`, default `http://localhost:4566`) and supplies dummy static credentials. ```go package main import ( "context" "encoding/json" "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/sns" "github.com/aws/aws-sdk-go-v2/service/sqs" sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types" ) 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, err := config.LoadDefaultConfig(ctx, config.WithRegion("us-east-1"), config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider("test", "test", "")), ) if err != nil { log.Fatalf("load config: %v", err) } snsClient := sns.NewFromConfig(cfg, func(o *sns.Options) { o.BaseEndpoint = aws.String(url) }) sqsClient := sqs.NewFromConfig(cfg, func(o *sqs.Options) { o.BaseEndpoint = aws.String(url) }) topic, _ := snsClient.CreateTopic(ctx, &sns.CreateTopicInput{Name: aws.String("notify")}) topicArn := aws.ToString(topic.TopicArn) // Two target queues, each subscribed (auto-confirmed). urlA, arnA := makeQueue(ctx, sqsClient, "q-a") urlB, arnB := makeQueue(ctx, sqsClient, "q-b") subscribe(ctx, snsClient, topicArn, arnA) subscribe(ctx, snsClient, topicArn, arnB) // One Publish fans out to both queues with a shared MessageId. pub, err := snsClient.Publish(ctx, &sns.PublishInput{ TopicArn: aws.String(topicArn), Message: aws.String("hello fan-out"), }) if err != nil { log.Fatalf("Publish: %v", err) } msgID := aws.ToString(pub.MessageId) fmt.Printf("Publish: MessageId=%s\n", msgID) for name, qURL := range map[string]string{"q-a": urlA, "q-b": urlB} { env := receiveEnvelope(ctx, sqsClient, qURL) if env.MessageId != msgID { log.Fatalf("FAIL: %s MessageId=%q != publish %q", name, env.MessageId, msgID) } fmt.Printf("%s received the publish (MessageId=%s)\n", name, env.MessageId) } } func makeQueue(ctx context.Context, c *sqs.Client, name string) (url, arn string) { created, _ := c.CreateQueue(ctx, &sqs.CreateQueueInput{QueueName: aws.String(name)}) url = aws.ToString(created.QueueUrl) out, _ := c.GetQueueAttributes(ctx, &sqs.GetQueueAttributesInput{ QueueUrl: aws.String(url), AttributeNames: []sqstypes.QueueAttributeName{sqstypes.QueueAttributeNameQueueArn}, }) return url, out.Attributes[string(sqstypes.QueueAttributeNameQueueArn)] } func subscribe(ctx context.Context, c *sns.Client, topicArn, queueArn string) { if _, err := c.Subscribe(ctx, &sns.SubscribeInput{ TopicArn: aws.String(topicArn), Protocol: aws.String("sqs"), Endpoint: aws.String(queueArn), ReturnSubscriptionArn: true, }); err != nil { log.Fatalf("Subscribe: %v", err) } } type notification struct{ MessageId, Message string } func receiveEnvelope(ctx context.Context, c *sqs.Client, queueURL string) notification { recv, _ := c.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{ QueueUrl: aws.String(queueURL), WaitTimeSeconds: 5, MaxNumberOfMessages: 1, }) var env notification _ = json.Unmarshal([]byte(aws.ToString(recv.Messages[0].Body)), &env) return env } ``` ```python import json import os import boto3 def make(service: str): return boto3.client( service, 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", ) def make_queue(sqs, name: str): url = sqs.create_queue(QueueName=name)["QueueUrl"] arn = sqs.get_queue_attributes(QueueUrl=url, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"] return url, arn def main() -> None: sns = make("sns") sqs = make("sqs") topic_arn = sns.create_topic(Name="notify")["TopicArn"] # Two target queues, each subscribed (auto-confirmed). url_a, arn_a = make_queue(sqs, "q-a") url_b, arn_b = make_queue(sqs, "q-b") for arn in (arn_a, arn_b): sns.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=arn, ReturnSubscriptionArn=True) # One Publish fans out to both queues with a shared MessageId. msg_id = sns.publish(TopicArn=topic_arn, Message="hello fan-out")["MessageId"] print(f"Publish -> MessageId={msg_id}") for name, url in (("q-a", url_a), ("q-b", url_b)): recv = sqs.receive_message(QueueUrl=url, WaitTimeSeconds=5, MaxNumberOfMessages=1) env = json.loads(recv["Messages"][0]["Body"]) assert env["MessageId"] == msg_id # same MessageId across deliveries print(f"{name} received the publish (MessageId={env['MessageId']})") if __name__ == "__main__": main() ``` ```java import java.net.URI; import java.util.List; import java.util.Map; 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.sns.SnsClient; import software.amazon.awssdk.services.sqs.SqsClient; import software.amazon.awssdk.services.sqs.model.QueueAttributeName; public final class Main { public static void main(String[] args) { String url = System.getenv().getOrDefault("KUBEMQ_AWS_URL", "http://localhost:4566"); StaticCredentialsProvider creds = StaticCredentialsProvider.create( AwsBasicCredentials.create("test", "test")); try (SnsClient sns = SnsClient.builder().endpointOverride(URI.create(url)) .region(Region.US_EAST_1).credentialsProvider(creds).build(); SqsClient sqs = SqsClient.builder().endpointOverride(URI.create(url)) .region(Region.US_EAST_1).credentialsProvider(creds).build()) { String topicArn = sns.createTopic(b -> b.name("notify")).topicArn(); // Two target queues, each subscribed (auto-confirmed). Map queues = Map.of("q-a", "", "q-b", ""); for (String name : List.of("q-a", "q-b")) { String queueUrl = sqs.createQueue(b -> b.queueName(name)).queueUrl(); String queueArn = sqs.getQueueAttributes(b -> b .queueUrl(queueUrl).attributeNames(QueueAttributeName.QUEUE_ARN)) .attributes().get(QueueAttributeName.QUEUE_ARN); sns.subscribe(b -> b.topicArn(topicArn).protocol("sqs").endpoint(queueArn) .returnSubscriptionArn(true)); queues = new java.util.HashMap<>(queues); queues.put(name, queueUrl); } // One Publish fans out to both queues with a shared MessageId. String messageId = sns.publish(b -> b.topicArn(topicArn).message("hello fan-out")) .messageId(); System.out.println("Publish -> MessageId=" + messageId); for (Map.Entry e : queues.entrySet()) { var recv = sqs.receiveMessage(b -> b.queueUrl(e.getValue()) .waitTimeSeconds(5).maxNumberOfMessages(1)); System.out.println(e.getKey() + " received: " + recv.messages().get(0).body()); } } } } ``` ```typescript import { SNSClient, CreateTopicCommand, SubscribeCommand, PublishCommand } from "@aws-sdk/client-sns"; import { SQSClient, CreateQueueCommand, GetQueueAttributesCommand, ReceiveMessageCommand, } from "@aws-sdk/client-sqs"; const url = process.env["KUBEMQ_AWS_URL"] ?? "http://localhost:4566"; const opts = { endpoint: url, region: "us-east-1", credentials: { accessKeyId: "test", secretAccessKey: "test" } }; const sns = new SNSClient(opts); const sqs = new SQSClient(opts); async function makeQueue(name: string): Promise<{ url: string; arn: string }> { const url = (await sqs.send(new CreateQueueCommand({ QueueName: name }))).QueueUrl!; const arn = ( await sqs.send(new GetQueueAttributesCommand({ QueueUrl: url, AttributeNames: ["QueueArn"] })) ).Attributes!["QueueArn"]!; return { url, arn }; } async function main(): Promise { const topicArn = (await sns.send(new CreateTopicCommand({ Name: "notify" }))).TopicArn!; // Two target queues, each subscribed (auto-confirmed). const a = await makeQueue("q-a"); const b = await makeQueue("q-b"); for (const q of [a, b]) { await sns.send(new SubscribeCommand({ TopicArn: topicArn, Protocol: "sqs", Endpoint: q.arn, ReturnSubscriptionArn: true })); } // One Publish fans out to both queues with a shared MessageId. const msgId = (await sns.send(new PublishCommand({ TopicArn: topicArn, Message: "hello fan-out" }))).MessageId; console.log(`Publish -> MessageId=${msgId}`); for (const [name, q] of [["q-a", a], ["q-b", b]] as const) { const recv = await sqs.send(new ReceiveMessageCommand({ QueueUrl: q.url, WaitTimeSeconds: 5, MaxNumberOfMessages: 1 })); const env = JSON.parse(recv.Messages![0].Body!); console.log(`${name} received the publish (MessageId=${env.MessageId})`); } } main().catch((err) => { console.error(err); process.exit(1); }); ``` ```csharp using System.Text.Json; using Amazon.Runtime; using Amazon.SimpleNotificationService; using Amazon.SimpleNotificationService.Model; using Amazon.SQS; using Amazon.SQS.Model; var url = Environment.GetEnvironmentVariable("KUBEMQ_AWS_URL") ?? "http://localhost:4566"; var creds = new BasicAWSCredentials("test", "test"); using var sns = new AmazonSimpleNotificationServiceClient(creds, new AmazonSimpleNotificationServiceConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" }); using var sqs = new AmazonSQSClient(creds, new AmazonSQSConfig { ServiceURL = url, AuthenticationRegion = "us-east-1" }); var topicArn = (await sns.CreateTopicAsync(new CreateTopicRequest { Name = "notify" })).TopicArn; async Task<(string url, string arn)> MakeQueue(string name) { var queueUrl = (await sqs.CreateQueueAsync(new CreateQueueRequest { QueueName = name })).QueueUrl; var arn = (await sqs.GetQueueAttributesAsync(new GetQueueAttributesRequest { QueueUrl = queueUrl, AttributeNames = ["QueueArn"], })).QueueARN; return (queueUrl, arn); } // Two target queues, each subscribed (auto-confirmed). var a = await MakeQueue("q-a"); var b = await MakeQueue("q-b"); foreach (var q in new[] { a, b }) { await sns.SubscribeAsync(new SubscribeRequest { TopicArn = topicArn, Protocol = "sqs", Endpoint = q.arn, ReturnSubscriptionArn = true, }); } // One Publish fans out to both queues with a shared MessageId. var msgId = (await sns.PublishAsync(new PublishRequest { TopicArn = topicArn, Message = "hello fan-out" })).MessageId; Console.WriteLine($"Publish -> MessageId={msgId}"); foreach (var (name, q) in new[] { ("q-a", a), ("q-b", b) }) { var recv = await sqs.ReceiveMessageAsync(new ReceiveMessageRequest { QueueUrl = q.url, WaitTimeSeconds = 5, MaxNumberOfMessages = 1, }); var env = JsonDocument.Parse(recv.Messages[0].Body).RootElement; Console.WriteLine($"{name} received the publish (MessageId={env.GetProperty("MessageId")})"); } ``` ```ruby # frozen_string_literal: true require "aws-sdk-sns" require "aws-sdk-sqs" require "json" Aws::SQS::Client.remove_plugin(Aws::SQS::Plugins::QueueUrls) url = ENV.fetch("KUBEMQ_AWS_URL", "http://localhost:4566") opts = { endpoint: url, region: "us-east-1", access_key_id: "test", secret_access_key: "test" } sns = Aws::SNS::Client.new(opts) sqs = Aws::SQS::Client.new(opts) topic_arn = sns.create_topic(name: "notify").topic_arn # Two target queues, each subscribed (auto-confirmed). def make_queue(sqs, name) url = sqs.create_queue(queue_name: name).queue_url arn = sqs.get_queue_attributes(queue_url: url, attribute_names: ["QueueArn"]).attributes["QueueArn"] [url, arn] end url_a, arn_a = make_queue(sqs, "q-a") url_b, arn_b = make_queue(sqs, "q-b") [arn_a, arn_b].each do |arn| sns.subscribe(topic_arn: topic_arn, protocol: "sqs", endpoint: arn, return_subscription_arn: true) end # One Publish fans out to both queues with a shared MessageId. msg_id = sns.publish(topic_arn: topic_arn, message: "hello fan-out").message_id puts "Publish -> MessageId=#{msg_id}" { "q-a" => url_a, "q-b" => url_b }.each do |name, queue_url| recv = sqs.receive_message(queue_url: queue_url, wait_time_seconds: 5, max_number_of_messages: 1) env = JSON.parse(recv.messages.first.body) puts "#{name} received the publish (MessageId=#{env['MessageId']})" end ``` ```rust use aws_config::{BehaviorVersion, Region}; use aws_credential_types::Credentials; use std::error::Error; async fn config() -> aws_config::SdkConfig { let url = std::env::var("KUBEMQ_AWS_URL").unwrap_or_else(|_| "http://localhost:4566".into()); 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 } async fn make_queue(sqs: &aws_sdk_sqs::Client, name: &str) -> (String, String) { let url = sqs.create_queue().queue_name(name).send().await.unwrap().queue_url.unwrap(); let arn = sqs .get_queue_attributes() .queue_url(&url) .attribute_names(aws_sdk_sqs::types::QueueAttributeName::QueueArn) .send() .await .unwrap() .attributes .and_then(|a| a.get(&aws_sdk_sqs::types::QueueAttributeName::QueueArn).cloned()) .unwrap(); (url, arn) } #[tokio::main] async fn main() -> Result<(), Box> { let conf = config().await; let sns = aws_sdk_sns::Client::new(&conf); let sqs = aws_sdk_sqs::Client::new(&conf); let topic_arn = sns.create_topic().name("notify").send().await?.topic_arn.unwrap(); // Two target queues, each subscribed (auto-confirmed). let (url_a, arn_a) = make_queue(&sqs, "q-a").await; let (url_b, arn_b) = make_queue(&sqs, "q-b").await; for arn in [arn_a, arn_b] { sns.subscribe() .topic_arn(&topic_arn) .protocol("sqs") .endpoint(arn) .return_subscription_arn(true) .send() .await?; } // One Publish fans out to both queues with a shared MessageId. let msg_id = sns.publish().topic_arn(&topic_arn).message("hello fan-out").send().await? .message_id.unwrap_or_default(); println!("Publish -> MessageId={msg_id}"); for (name, url) in [("q-a", url_a), ("q-b", url_b)] { let recv = sqs.receive_message().queue_url(&url).wait_time_seconds(5).max_number_of_messages(1) .send().await?; println!("{name} received: {}", recv.messages().first().and_then(|m| m.body()).unwrap_or_default()); } Ok(()) } ``` ## Filtered fan-out [#filtered-fan-out] Attach a `FilterPolicy` (on the `MessageAttributes` scope) to a subscription so it receives only the publishes it cares about. A matching publish is delivered; a non-matching one is suppressed — and a publish that matches **no** subscription still succeeds. ```python # Only deliver publishes whose "eventType" attribute is "order". sns.subscribe( TopicArn=topic_arn, Protocol="sqs", Endpoint=queue_arn, Attributes={ "FilterPolicy": json.dumps({"eventType": ["order"]}), "FilterPolicyScope": "MessageAttributes", }, ReturnSubscriptionArn=True, ) sns.publish(TopicArn=topic_arn, Message="an order event", MessageAttributes={"eventType": {"DataType": "String", "StringValue": "order"}}) sns.publish(TopicArn=topic_arn, Message="a metric event", MessageAttributes={"eventType": {"DataType": "String", "StringValue": "metric"}}) # Only "an order event" is delivered; the metric publish is suppressed. ``` **Filtering works on `MessageAttributes` scope only.** Setting `FilterPolicyScope = MessageBody` is rejected with `InvalidParameter`. Put the values you filter on into message attributes, not the body. ## Raw vs enveloped delivery [#raw-vs-enveloped-delivery] `RawMessageDelivery` controls the delivered shape per subscription: * **Enveloped (default):** the SQS message body is the SNS `Notification` JSON (`MessageId`, `TopicArn`, `Message`, `UnsubscribeURL`, `MessageAttributes`, `SignatureVersion: "1"`, and an empty `Signature`). * **Raw:** for SQS, the bare body plus `sns_topic_arn` / `sns_subject` tags and the attribute codec. ```python # One subscription raw, one enveloped, on the same topic. sns.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=raw_arn, Attributes={"RawMessageDelivery": "true"}, ReturnSubscriptionArn=True) sns.subscribe(TopicArn=topic_arn, Protocol="sqs", Endpoint=env_arn, ReturnSubscriptionArn=True) # default = enveloped ``` **Delivered SNS notifications are unsigned.** The `Notification` envelope carries `SignatureVersion: "1"` but its `Signature` and `SigningCertURL` are **empty** — a receiver cannot verify the message signature. ## HTTP/HTTPS webhooks [#httphttps-webhooks] A topic can also fan out to `http` / `https` endpoints. Such a subscription goes **pending** until the subscriber calls `ConfirmSubscription` (the only SigV4-exempt action, so the embedded `SubscribeURL` works as an unsigned GET). Once confirmed, deliveries run through an in-memory engine: 8 workers, a bounded job queue, a default 51-attempt retry schedule (overridable by a stored `DeliveryPolicy`), a per-endpoint circuit breaker that opens after 5 consecutive failures (30 s), and a redrive to the subscription's `RedrivePolicy` DLQ on exhaustion. Raw HTTP delivery maps attributes to `x-amz-sns-attr-{name}` headers. See the [reliability guide](/connectors/aws/how-to/reliability) and the [SNS fan-out guide](/connectors/aws/how-to/sns-fan-out). **SNS HTTP delivery state is in-memory on the publishing node.** A node restart loses pending retries, and the bounded job queue drops on overflow. This is part of the node-local / sticky-load-balancer family of caveats. ## Related [#related]