Setting up Numaflow MonoVertex with SQS Source¶

This guide explains how to set up a Numaflow MonoVertex that reads from an AWS SQS queue.

Configuring Credentials to access AWS¶

There are couple of ways you could achieve this.

Creating AWS Credentials Secret¶

First, we need to create a Kubernetes secret to store AWS credentials securely and encode AWS Credentials.

# Encode your AWS credentials (replace with your actual credentials)
ACCESS_KEY_ID=$(echo -n "your-aws-access-key-id" | base64)
SECRET_ACCESS_KEY=$(echo -n "your-aws-secret-access-key" | base64)

kubectl create secret generic aws-secret \
--from-literal access-key-id=${ACCESS_KEY_ID} \
--from-literal secret-access-key=${SECRET_ACCESS_KEY}

IAM ROLE¶

Use an IAM role with the necessary permissions to access the SQS queue.
Ensure the IAM role of the pod has access to the SQS queue, and the SQS queue policy allows the IAM role to perform actions on the queue.
Attach the appropriate service account with the IAM role to the pod.

For more details, refer to the AWS documentation: https://docs.aws.amazon.com/eks/latest/userguide/pod-id-how-it-works.html

Create the numaflow pipeline¶

Create a file named sqs-pl.yaml with either of the following content.

YAML / key	Where	Meaning
`queueName`	source spec	Read this one queue
`queueNames`	source spec	Read these queues (XOR with `queueName`)
`queue_name`	system metadata, group `sqs`	Origin of this message (single- or multi-queue)
`queueName`	sink spec	Write dest (builtin SQS sink; unchanged by origin)

User metadata group sqs is producer or UDF attributes, not origin. See Source Queue Origin.

MonoVertex Specification¶

apiVersion: numaflow.numaproj.io/v1alpha1
kind: MonoVertex
metadata:
  name: sqs-reader
spec:
    source:
        sqs:
          queueName: "your-queue-name"        # Required: Name of your SQS queue
          awsRegion: "your-aws-region"        # Required: AWS region where queue is located
          queueOwnerAWSAccountID: "123456789012" # Required: AWS account ID of the queue owner
          # Optional configurations
          maxNumberOfMessages: 10             # Max messages per poll (1-10)
          visibilityTimeout: 30              # Visibility timeout in seconds
          waitTimeSeconds: 20                # Long polling wait time
          attributeNames:                    # SQS attributes to retrieve
            - All
          messageAttributeNames:             # Message attributes to retrieve
            - All
    sink:
        log: {}                             # Prints payload/headers, not system metadata
    limits:
        readBatchSize: 10
        bufferSize: 100
    scale:
        min: 1
        max: 5

Pipeline Specification¶

apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: sqs-source-pl
spec:
  vertices:
    - name: in
      scale:
        min: 1
      source:
        sqs:
          queueName: "queue-name"
          awsRegion: "us-west-2"
          queueOwnerAWSAccountID: "123456789012" # Required: AWS account ID of the queue owner
          # Optional configurations
          maxNumberOfMessages: 10            # Max messages per poll (1-10)
          visibilityTimeout: 30              # Visibility timeout in seconds
          waitTimeSeconds: 20                # Long polling wait time
          attributeNames:                    # SQS attributes to retrieve
            - All
          messageAttributeNames:             # Message attributes to retrieve
            - All
    - name: out
      scale:
        min: 1
      sink:
        log: {}
  edges:
    - from: in
      to: out

Multiple Queues¶

A single SQS source can consume from multiple queues in the same AWS account and region by using queueNames instead of queueName. queueNames is a comma-separated list; the two queue selector fields are mutually exclusive.

source:
  sqs:
    queueNames: "orders-queue,refunds-queue,replay-queue"
    awsRegion: "us-east-1"
    queueOwnerAWSAccountID: "111111111111"
    visibilityTimeout: 30
    maxNumberOfMessages: 10
    waitTimeSeconds: 20

Every listed queue uses the same:

AWS account and region;
pod credentials or assumeRole;
endpointUrl;
visibility and polling settings; and
system and message attribute selections.

The IAM identity must have access to every queue. If queues require different accounts, regions, credentials, or tuning, configure separate source vertices.

Messages from all configured queues are merged without a cross-queue ordering guarantee. Origin is on every SQS message; see Source Queue Origin.

Read and Failure Behavior¶

Every queue is polled concurrently, and a read returns as soon as any queue has messages. A slow or empty queue never holds back messages already received from the others. If a queue answers after the read has returned, its messages are held and served on the next read rather than waiting out their visibility timeout.

Receive failures are handled per queue. A failing queue does not discard messages already received from healthy queues, and a transient error (a throttle, for example) is retried on subsequent reads. Only when one queue fails ten consecutive receives does the source stop, so the pod restarts and every queue URL is resolved again. Until then, the remaining queues keep flowing.

Apply the Configuration¶

Apply the pipeline specification:

kubectl apply -f sqs-pl.yaml

Verify the Setup¶

Check that the MonoVertex is running:

kubectl get monovertex sqs-reader
kubectl get pods -l numaflow.numaproj.io/vertex-name=sqs-reader

Message Attributes and Headers¶

The SQS source propagates message attributes through the pipeline, making them available to UDFs and downstream sinks.

System Attributes¶

When attributeNames is configured, SQS system attributes (e.g., SentTimestamp, ApproximateReceiveCount, MessageGroupId, MessageDeduplicationId) are propagated as message headers. These are accessible in UDFs (e.g., datum.Headers() (Go), datum.headers (Python), etc.).

Custom Attributes¶

When messageAttributeNames is configured, user-defined message attributes are propagated as user metadata under the sqs namespace. These are accessible in UDFs via the metadata API.

Source Queue Origin¶

Every message from an SQS source includes the queue it was read from as system metadata (not headers, not user metadata). This is stamped for both queueName and queueNames:

Group: sqs
Key: queue_name

Transformers and map UDFs copy parent system metadata, so origin survives downstream. The builtin log sink does not print system metadata; use a UDF or UDSink to see it. The builtin SQS sink destination is still that vertex's YAML queueName.

origin = datum.system_metadata.value("sqs", "queue_name")

The exact accessor depends on the SDK; the value lives in the Datum/request system metadata under group sqs, key queue_name.

Example: Accessing SQS Attributes in a UDF¶

Headers below; origin is system metadata (see Source Queue Origin).

Go SDK:

func handler(ctx context.Context, keys []string, datum functionsdk.Datum) functionsdk.Messages {
    // Access system attributes via headers
    headers := datum.Headers()
    sentTimestamp := headers["SentTimestamp"]
    messageGroupId := headers["MessageGroupId"]

    // Process message...
    return functionsdk.MessagesBuilder().Append(datum.Value())
}

Python SDK:

def handler(keys: list[str], datum: Datum) -> Messages:
    # Access system attributes via headers
    headers = datum.headers
    sent_timestamp = headers.get("SentTimestamp")
    message_group_id = headers.get("MessageGroupId")

    # Process message...
    return Messages(Message(datum.value))

FIFO Queue Support¶

For FIFO queues, the MessageGroupId and MessageDeduplicationId attributes are automatically propagated through the pipeline. When using an SQS sink, these values are passed to the destination queue to maintain FIFO ordering and deduplication.

Troubleshooting¶

Check the pods logs for any errors:

kubectl logs -l numaflow.numaproj.io/vertex-name=sqs-reader

Verify the secret exists:
```
kubectl get secret aws-secret
```
Ensure the SQS queue exists and is accessible with the provided credentials