feat: add sqs source #2355

cosmic-chichu · 2025-01-23T03:07:00Z

resolves #2367

Add SQS source
Add unit tests for sqs source
Add end to end test for sqs source

Signed-off-by: Vigith Maurice <[email protected]>

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

rust/extns/numaflow-sqs/Cargo.toml

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

BulkBeing · 2025-01-23T03:45:31Z

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

+use std::collections::HashMap;
+use std::time::Duration;
+use tokio::sync::{mpsc, oneshot};
+use tokio::time::Instant;
+
+use aws_config::{meta::region::RegionProviderChain, BehaviorVersion};
+use aws_sdk_sqs::config::Region;
+use aws_sdk_sqs::types::{MessageSystemAttributeName, QueueAttributeName};
+use aws_sdk_sqs::{Client};
+use bytes::Bytes;
+use chrono::{DateTime, TimeZone, Utc};
+
+use crate::{Error, Result};


Group imports in std-external-crate order https://rust-lang.github.io/rustfmt/?version=master&search=#StdExternalCrate%5C%3A
We try to follow this for new changes.

I have added --config group_imports=StdExternalCrate to my Rust Rover rustfmt arguments. Should we add a rustfmt.toml?

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

rust/numaflow-core/Cargo.toml

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

vigith · 2025-01-25T03:42:28Z

please resolve conflicts

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

codecov · 2025-01-27T10:43:04Z

Codecov Report

Attention: Patch coverage is 89.29987% with 81 lines in your changes missing coverage. Please review.

Project coverage is 69.87%. Comparing base (8e9bafb) to head (d0e426e).

Files with missing lines	Patch %	Lines
rust/extns/numaflow-sqs/src/source.rs	88.31%	50 Missing ⚠️
rust/numaflow-core/src/source/sqs.rs	88.21%	31 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2355      +/-   ##
==========================================
+ Coverage   69.84%   69.87%   +0.03%     
==========================================
  Files         361      364       +3     
  Lines       49935    50823     +888     
==========================================
+ Hits        34878    35515     +637     
- Misses      13979    14229     +250     
- Partials     1078     1079       +1

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

vigith · 2025-01-28T01:37:28Z

rust/extns/numaflow-sqs/Cargo.toml

+tonic = "0.12.3"
+prost = "0.11.9"
+thiserror = "1.0.69"
+http = "1.2.0"


do we need http?

@cosmic-chichu You can use cargo-udeps to identify crates that are not used.

# Install cargo install cargo-udeps # Go to the crates directory and run it. It needs nightly compiler cd rust/extns/numaflow-sqs cargo +nightly udeps --all-targets

vigith · 2025-01-28T01:37:42Z

rust/extns/numaflow-sqs/Cargo.toml

+chrono = "0.4.38"
+tonic = "0.12.3"
+prost = "0.11.9"
+thiserror = "1.0.69"


vigith · 2025-01-28T01:40:20Z

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

+#[derive(thiserror::Error, Debug)]
+pub enum Error {
+    #[error("Failed with SQS error - {0}")]
+    SQS(aws_sdk_sqs::Error),


to handle this error in the core, we will have to import aws_sdk_sqs, right? so convert this to numaflow error.

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

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

BulkBeing · 2025-01-28T06:49:02Z

@cosmic-chichu Run clippy for the crates in which you made changes and make the changes suggested by clippy:

cd rust/extns/numaflow-sqs

cargo clippy --tests --all-features --no-deps

yhl25 · 2025-01-28T07:08:59Z

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

+pub struct SQSSourceConfig {
+    pub region: String,
+    pub queue_name: String,


we might have to enhance it with more sqs source related configs like visibility timeout, poll timeout etc.

Lets add support for all the configurations that is supported by SQS Client.

yhl25 · 2025-01-28T07:15:23Z

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

+/// Used to initialize the SQS client with region and queue settings.
+/// Implements serde::Deserialize to support loading from configuration files.


we might need to support different ways to authenticate (example role-based)

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

yhl25 · 2025-01-28T07:24:40Z

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

+            respond_to: tx,
+            queue_name: self.queue_name.clone(),
+            count: self.batch_size as i32,
+            timeout_at: Instant::now() + self.timeout,


Suggested change

timeout_at: Instant::now() + self.timeout,

timeout_at: start + self.timeout,

yhl25 · 2025-01-28T07:46:03Z

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

+                timeout_at,
+            } => {
+                let messages = self.get_messages(queue_name, count, timeout_at).await;
+                let _ = respond_to.send(messages);


use expect?

yhl25 · 2025-01-28T07:46:18Z

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

+                }
+                _ => {
+                    tracing::error!("Unsupported attribute name");
+                    let _ = respond_to


use expect?

yhl25 · 2025-01-28T07:48:31Z

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

+                let status = self.delete_messages(queue_name, offsets).await;
+                let _ = respond_to.send(status);
+            }
+            SQSActorMessage::GetQueueAttributes {


Can we avoid this actor message and move this functionality inside pending?

yhl25 · 2025-01-28T07:49:47Z

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

+    GetQueueAttributes {
+        respond_to: oneshot::Sender<Result<Option<usize>>>,
+        queue_name: String,
+        attribute_name: QueueAttributeName,
+    },


make this pending?

yhl25 · 2025-01-28T07:50:58Z

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

+    }
+
+    /// Retrieves messages from SQS with timeout and batching.
+    ///


make queue static to avoid dynamic queue url creation and to avoid caching?

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

vigith · 2025-02-02T19:30:02Z

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

            event_time: message.event_time,
+            watermark: None,


why not set Watermark based of SentTimestamp in the Attributes

vigith and others added 2 commits January 16, 2025 16:14

chore: add numaflow-sqs extn

d1d83a7

Signed-off-by: Vigith Maurice <[email protected]>

feat: add sqs source

551459c

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

cosmic-chichu requested review from whynowy and vigith as code owners January 23, 2025 03:07

vigith requested review from BulkBeing and yhl25 January 23, 2025 03:08

BulkBeing reviewed Jan 23, 2025

View reviewed changes

rust/extns/numaflow-sqs/Cargo.toml Outdated Show resolved Hide resolved

BulkBeing reviewed Jan 23, 2025

View reviewed changes

rust/extns/numaflow-sqs/src/lib.rs Outdated Show resolved Hide resolved

BulkBeing reviewed Jan 23, 2025

View reviewed changes

rust/extns/numaflow-sqs/src/source.rs Outdated Show resolved Hide resolved

BulkBeing reviewed Jan 23, 2025

View reviewed changes

rust/numaflow-core/Cargo.toml Outdated Show resolved Hide resolved

BulkBeing reviewed Jan 23, 2025

View reviewed changes

rust/numaflow-core/src/source/sqs.rs Show resolved Hide resolved

cosmic-chichu added 4 commits January 27, 2025 00:13

fix: add tests and address review comments

590905e

fix: merge conflicts

3f592b1

fix: integration test

8bb5577

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

fix: fmt

90c9a23

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

cosmic-chichu requested a review from BulkBeing January 27, 2025 11:07

vigith reviewed Jan 28, 2025

View reviewed changes

rust/extns/numaflow-sqs/src/source.rs Outdated Show resolved Hide resolved

yhl25 reviewed Jan 28, 2025

View reviewed changes

yhl25 assigned cosmic-chichu Jan 31, 2025

yhl25 added this to the 1.5 milestone Jan 31, 2025

cosmic-chichu added 3 commits February 1, 2025 00:04

fix: addressed review comments

630ed9b

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

fix: merge conflicts

ed3a1b8

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

fix: sqs source

d0e426e

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

vigith reviewed Feb 2, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add sqs source #2355

feat: add sqs source #2355

cosmic-chichu commented Jan 23, 2025 •

edited by vigith

Loading

BulkBeing Jan 23, 2025

cosmic-chichu Jan 27, 2025

vigith commented Jan 25, 2025

codecov bot commented Jan 27, 2025 •

edited

Loading

vigith Jan 28, 2025

BulkBeing Jan 28, 2025

vigith Jan 28, 2025

vigith Jan 28, 2025

BulkBeing commented Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

yhl25 Jan 28, 2025

vigith Feb 2, 2025

		/// Used to initialize the SQS client with region and queue settings.
		/// Implements serde::Deserialize to support loading from configuration files.

	timeout_at: Instant::now() + self.timeout,
	timeout_at: start + self.timeout,

feat: add sqs source #2355

Are you sure you want to change the base?

feat: add sqs source #2355

Conversation

cosmic-chichu commented Jan 23, 2025 • edited by vigith Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vigith commented Jan 25, 2025

codecov bot commented Jan 27, 2025 • edited Loading

Codecov Report

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BulkBeing commented Jan 28, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cosmic-chichu commented Jan 23, 2025 •

edited by vigith

Loading

codecov bot commented Jan 27, 2025 •

edited

Loading