feat: SQS System Attributes and Custom Attributes Propagation #3095

cosmic-chichu · 2025-12-08T14:41:57Z

Summary

Enables propagation of SQS attributes through Numaflow pipelines, supporting FIFO queue semantics (MessageGroupId, MessageDeduplicationId) and user-defined message attributes.

Changes

SQS Source: Attribute extraction
Renamed SqsMessage fields to system_attributes (SQS system attrs like SentTimestamp, MessageGroupId) and custom_attributes (user-defined message attributes). System attributes become message headers; custom attributes are stored in metadata.user_metadata["sqs"].
SQS Sink: FIFO queue support
Added headers field to SqsSinkMessage. The sink reads MessageGroupId, MessageDeduplicationId, and DelaySeconds from headers and applies them to the SQS SendMessageBatch request.
numaflow-core integration
Source maps system_attributes → Message.headers and custom_attributes → Message.metadata. Sink merges headers with metadata.user_metadata["sqs"], allowing UDFs/transformers to set or override FIFO attributes.
Improved error messages
Changed Error::Sqs to store extracted error details instead of opaque AWS SDK error. Errors now display as "ErrorCode: Error message" instead of "unhandled error (CODE)".
Documentation
Added sink documentation (docs/user-guide/sinks/sqs.md), updated source docs with attribute configuration examples, and added SQS example to message headers concept page.

Signed-off-by: Shrivardhan Rao <[email protected]>

codecov · 2025-12-08T14:56:51Z

Codecov Report

❌ Patch coverage is 97.38806% with 7 lines in your changes missing coverage. Please review.
✅ Project coverage is 79.75%. Comparing base (7f28dc7) to head (70fd91e).

Files with missing lines	Patch %	Lines
rust/extns/numaflow-sqs/src/sink.rs	92.68%	3 Missing ⚠️
rust/extns/numaflow-sqs/src/lib.rs	88.88%	2 Missing ⚠️
rust/extns/numaflow-sqs/src/source.rs	90.00%	2 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #3095      +/-   ##
==========================================
+ Coverage   79.73%   79.75%   +0.01%     
==========================================
  Files         291      291              
  Lines       65143    65376     +233     
==========================================
+ Hits        51941    52138     +197     
- Misses      12649    12690      +41     
+ Partials      553      548       -5

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Signed-off-by: Shrivardhan Rao <[email protected]>

…aproj/numaflow into dev-nightly-feat/sqs-headers

Signed-off-by: Shrivardhan Rao <[email protected]>

…aproj/numaflow into dev-nightly-feat/sqs-headers

Signed-off-by: Shrivardhan Rao <[email protected]>

KeranYang · 2026-01-07T16:11:19Z

docs/core-concepts/message-headers.md

  In `pkg/sources/http/http.go`, HTTP headers from incoming requests are added to the message headers. This means any custom or standard HTTP header is available throughout the pipeline.

+- **SQS Source Example**  
+  When configured with `attributeNames`, SQS system attributes (e.g., `SentTimestamp`, `MessageGroupId`, `MessageDeduplicationId`) are copied into the Numaflow message headers. User-defined message attributes (configured via `messageAttributeNames`) are propagated as metadata under the `sqs` namespace.


I see the following statement above.

Message headers are immutable and cannot be manipulated through the SDKs.

UDFs can read message headers to enrich processing logic, but they cannot modify or add new headers.

With this PR, it looks UDF can now add headers. Do we need to update the doc to be consistent?

this PR enables adding custom key/values to the metadata name spaced under sqs. Headers are still immutable.

KeranYang · 2026-01-07T16:12:32Z

docs/user-guide/sinks/sqs.md

+
+### Setting Headers via Metadata
+
+User-defined functions can set SQS headers by writing to the `sqs` metadata namespace. These values are merged into the message headers at the sink, allowing UDFs to control FIFO ordering or set delay seconds.


Is it via SDK? Can you give an example on how to set?

KeranYang · 2026-01-07T16:23:35Z

rust/numaflow-core/src/sinker/sink/sqs.rs

+        if let Some(metadata) = &msg.metadata {
+            if let Some(sqs_meta) = metadata.user_metadata.get("sqs") {
+                for (k, v) in &sqs_meta.key_value {
+                    if let Ok(val_str) = String::from_utf8(v.to_vec()) {


Invalid UTF-8 will be silently ignored. Should we match and print warning for Err case?

KeranYang · 2026-01-07T16:25:40Z

rust/extns/numaflow-sqs/src/sink.rs

-                })?;
+                .message_body(String::from_utf8_lossy(&message.message_body).to_string());
+
+            if let Some(delay) = message.headers.get("DelaySeconds") {


Invalid DelaySeconds are silently ignored, please match and handle error case. Do we have value validation on paved path end? I imagine DelaySeconds cannot be higher than certain threshold.

KeranYang · 2026-01-07T16:30:50Z

rust/extns/numaflow-sqs/src/source.rs

+                            sqs_attrs.insert(k.clone(), val.clone().into_bytes());
+                        }
+                    }
+                    custom_attributes.insert("sqs".to_string(), sqs_attrs);


Can we make sqs a const in sqs crate and use the const in numaflow-core?

KeranYang · 2026-01-07T16:32:03Z

rust/extns/numaflow-sqs/src/lib.rs


+/// Extracts a user-friendly error message from AWS SDK errors.
+/// Returns format: "ErrorCode: Error message" or falls back to string representation.
+pub fn extract_aws_error<E, R>(err: &aws_sdk_sqs::error::SdkError<E, R>) -> String


Suggested change

pub fn extract_aws_error<E, R>(err: &aws_sdk_sqs::error::SdkError<E, R>) -> String

pub(crate) fn extract_aws_error<E, R>(err: &aws_sdk_sqs::error::SdkError<E, R>) -> String

docs/user-guide/sources/sqs.md

vigith · 2026-01-07T04:32:03Z

docs/user-guide/sinks/sqs.md

add to mkdocs.yml

vigith · 2026-01-07T04:33:53Z

rust/extns/numaflow-sqs/src/lib.rs

-    #[error("Failed with SQS error - {0}")]
-    Sqs(#[from] aws_sdk_sqs::Error),
+    #[error("{0}")]
+    Sqs(String),


why change this?

rust/extns/numaflow-sqs/src/sink.rs

Co-authored-by: Vigith Maurice <[email protected]>

cosmic-chichu added 2 commits December 8, 2025 06:34

feat: add control headers for sqs

f7b674b

Signed-off-by: Shrivardhan Rao <[email protected]>

fix: lint

67a1cfb

Signed-off-by: Shrivardhan Rao <[email protected]>

cosmic-chichu added 14 commits December 14, 2025 18:59

Merge branch 'main' into dev-nightly-feat/sqs-headers

d2e4c35

Merge branch 'main' into dev-nightly-feat/sqs-headers

53c049f

fix: merge headers into metadta

ea6bf1d

Signed-off-by: Shrivardhan Rao <[email protected]>

Merge branch 'dev-nightly-feat/sqs-headers' of https://github.com/num…

118d2a8

…aproj/numaflow into dev-nightly-feat/sqs-headers

Merge branch 'main' into dev-nightly-feat/sqs-headers

d74fd9c

fix: add debug log

b5a1672

Signed-off-by: Shrivardhan Rao <[email protected]>

Merge branch 'main' into dev-nightly-feat/sqs-headers

7c02d4e

Merge branch 'dev-nightly-feat/sqs-headers' of https://github.com/num…

ecde562

…aproj/numaflow into dev-nightly-feat/sqs-headers

feat: copy custom message attributes into metadata

2ebd37b

Signed-off-by: Shrivardhan Rao <[email protected]>

fix: update SqsMessage

7bab509

Signed-off-by: Shrivardhan Rao <[email protected]>

fix: lint

3e12848

Signed-off-by: Shrivardhan Rao <[email protected]>

fix: update errors and add docs

3048f8e

Signed-off-by: Shrivardhan Rao <[email protected]>

fix: add sink docs

daf5fab

Signed-off-by: Shrivardhan Rao <[email protected]>

fix: lint

7a773f6

Signed-off-by: Shrivardhan Rao <[email protected]>

cosmic-chichu marked this pull request as ready for review January 7, 2026 00:39

cosmic-chichu requested review from vigith, whynowy and yhl25 as code owners January 7, 2026 00:39

cosmic-chichu changed the title ~~feat: add control headers for sqs~~ feat: SQS System Attributes and Custom Attributes Propagation Jan 7, 2026

fix: remove debug log

0f6058f

Signed-off-by: Shrivardhan Rao <[email protected]>

KeranYang reviewed Jan 7, 2026

View reviewed changes

vigith reviewed Jan 7, 2026

View reviewed changes

Update docs/user-guide/sources/sqs.md

70fd91e

Co-authored-by: Vigith Maurice <[email protected]>

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat: SQS System Attributes and Custom Attributes Propagation #3095

feat: SQS System Attributes and Custom Attributes Propagation #3095

Uh oh!

cosmic-chichu commented Dec 8, 2025 •

edited

Loading

Uh oh!

codecov bot commented Dec 8, 2025 •

edited

Loading

Uh oh!

KeranYang Jan 7, 2026

Uh oh!

cosmic-chichu Jan 7, 2026

Uh oh!

KeranYang Jan 7, 2026

Uh oh!

KeranYang Jan 7, 2026

Uh oh!

KeranYang Jan 7, 2026

Uh oh!

KeranYang Jan 7, 2026

Uh oh!

KeranYang Jan 7, 2026

Uh oh!

Uh oh!

vigith Jan 7, 2026

Uh oh!

vigith Jan 7, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants


		### Setting Headers via Metadata

		User-defined functions can set SQS headers by writing to the `sqs` metadata namespace. These values are merged into the message headers at the sink, allowing UDFs to control FIFO ordering or set delay seconds.

	pub fn extract_aws_error<E, R>(err: &aws_sdk_sqs::error::SdkError<E, R>) -> String
	pub(crate) fn extract_aws_error<E, R>(err: &aws_sdk_sqs::error::SdkError<E, R>) -> String

feat: SQS System Attributes and Custom Attributes Propagation #3095

Are you sure you want to change the base?

feat: SQS System Attributes and Custom Attributes Propagation #3095

Uh oh!

Conversation

cosmic-chichu commented Dec 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Changes

Uh oh!

codecov bot commented Dec 8, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

cosmic-chichu commented Dec 8, 2025 •

edited

Loading

codecov bot commented Dec 8, 2025 •

edited

Loading