pubsub: Enable sendAck and sendNack apis for AWS Pubsub #134

roseyang62 · 2025-11-07T21:46:07Z

Summary

Implemented doSendAcks() and doSendNacks() to support acknowledging and negatively acknowledging messages for AWS SQS.
- sendAck uses the SQS DeleteMessageBatch API to remove successfully processed messages.
- sendNack uses the SQS ChangeMessageVisibilityBatch API to make failed messages visible again for redelivery.
Added integration tests
Added AwsMatcherRelaxingTransformer to relax WireMock’s body matching for DeleteMessageBatch or ChangeMessageVisibilityBatch operations.
remove AckID interface

Some conventions to follow

add the module name as a prefix
- for example: add a prefix: docstore: for document store module, blobstore for Blob Store module
for a test only PR, add test:
for a perf improvement only PR, add perf:
for a refactoring only PR, add "refactor:"

codecov-commenter · 2025-11-07T21:51:53Z

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 83.32%. Comparing base (575f364) to head (e927a09).
⚠️ Report is 18 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main     #134      +/-   ##
============================================
+ Coverage     83.06%   83.32%   +0.25%     
  Complexity       36       36              
============================================
  Files           148      148              
  Lines          7635     7710      +75     
  Branches        888      901      +13     
============================================
+ Hits           6342     6424      +82     
+ Misses          859      853       -6     
+ Partials        434      433       -1

Flag	Coverage Δ
unittests	`83.32% <100.00%> (+0.25%)`	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ 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.

sandeepvinayak · 2025-11-10T22:43:54Z

pubsub/pubsub-aws/src/main/java/com/salesforce/multicloudj/pubsub/aws/AwsSubscription.java

+        List<String> receiptHandles = new ArrayList<>();
+        for (AckID ackID : ackIDs) {
+            if (ackID instanceof AwsAckID) {
+                receiptHandles.add(((AwsAckID) ackID).getReceiptHandle());
+            } else {
+                throw new IllegalArgumentException("Invalid AckID type: " + ackID.getClass());
+            }
+        }


under what scenario will this not be instance of AwsAckID ? this seems like internals to SDK and throwing IllegalArgumentException to user for something abstracted doesn't make much sense.

yep.. The instanceof check here doesn’t add any value. I removed it.

actually thinking aloud, why do we need AckID interface? AWS and GCP both have strings at AckIDs, then not sure if this interface provides any value.

I had a similar thought when I was working on the GCP implementation. this AckID interface doesn’t seem to add much value.
But I wasn’t sure how AWS or Ali drivers were handling their AckIDs, and I didn’t want to touch the driver-level architecture at that time.
I’m trying to replace it with strings now, and it should have no functional impact.

sandeepvinayak · 2025-11-10T22:47:18Z

pubsub/pubsub-aws/src/main/java/com/salesforce/multicloudj/pubsub/aws/AwsSubscription.java

+        // SQS supports max 10 messages per batch operation
+        for (int i = 0; i < receiptHandles.size(); i += 10) {
+            int endIndex = Math.min(i + 10, receiptHandles.size());
+            List<String> batch = receiptHandles.subList(i, endIndex);


why would we need additional memory when already have receipt handles ?

updated. Removed subList() and now use receiptHandles directly to avoid extra memory and simplify batching.

sandeepvinayak · 2025-11-10T22:47:45Z

pubsub/pubsub-aws/src/main/java/com/salesforce/multicloudj/pubsub/aws/AwsSubscription.java

+
+        List<String> receiptHandles = new ArrayList<>();
+        for (AckID ackID : ackIDs) {
+            if (ackID instanceof AwsAckID) {


sandeepvinayak · 2025-11-10T22:47:54Z

pubsub/pubsub-aws/src/main/java/com/salesforce/multicloudj/pubsub/aws/AwsSubscription.java

+
+        for (int i = 0; i < receiptHandles.size(); i += 10) {
+            int endIndex = Math.min(i + 10, receiptHandles.size());
+            List<String> batch = receiptHandles.subList(i, endIndex);


sandeepvinayak · 2025-11-10T22:48:19Z

pubsub/pubsub-aws/src/main/java/com/salesforce/multicloudj/pubsub/aws/AwsSubscription.java

+
+            if (!actualFailures.isEmpty()) {
+                BatchResultErrorEntry firstFailure = actualFailures.get(0);
+                throw new RuntimeException(


let's not throw RunTimeException

sandeepvinayak · 2025-11-10T22:50:35Z

...b-aws/src/test/java/com/salesforce/multicloudj/pubsub/aws/AwsMatcherRelaxingTransformer.java

+        List<ContentPattern<?>> bodyPatterns = stub.getRequest().getBodyPatterns();
+        if (bodyPatterns == null) {
+            return stub;
+        }
+
+        // Relax only batch operations
+        if (amzTarget.contains("DeleteMessageBatch")) {
+            bodyPatterns.clear();
+            bodyPatterns.add(new MatchesJsonPathPattern("$.Entries[*].ReceiptHandle"));
+        } 
+        else if (amzTarget.contains("ChangeMessageVisibilityBatch")) {
+            bodyPatterns.clear();
+            bodyPatterns.add(new MatchesJsonPathPattern("$.Entries[*].ReceiptHandle"));
+            bodyPatterns.add(new MatchesJsonPathPattern("$.Entries[*].VisibilityTimeout"));
+        }
+        else if (amzTarget.contains("SendMessageBatch")) {
+            // keep body integrity to preserve MD5 validation
+            // just verify structure, but don't clear existing patterns
+            bodyPatterns.add(new MatchesJsonPathPattern("$.Entries[*].MessageBody"));
+        }
+        // For ReceiveMessage — normally no body, skip
+        else if (amzTarget.contains("ReceiveMessage")) {
+            return stub;
+        }


why are we doing these special handling ?

these batch operations (DeleteMessageBatch, ChangeMessageVisibilityBatch, SendMessageBatch) need relaxed matching because their request bodies include dynamic fields like receipt handles or timestamps that vary between replay runs. Without this special handling, our recorded stubs would fail to match during replay.

Not sure how's that happening. Receipt handle is from server side, the client don't create it. Where does it getting randomized?

batch size changes during replay can cause WireMock to match a different ReceiveMessage stub and return a different AWS-recorded ReceiptHandle. This mismatch causes the replay phase cannot match the recorded mappings.

During record:
AWS returned multiple different ReceiptHandles across multiple ReceiveMessage calls:

Call batchSize Mapping matched ReceiptHandle returned

1 1 mapping A A1

2 2 mapping B B1

3 1 mapping C C1

During replay:
If the batchSize changes (e.g., from 2 → 1), the second ReceiveMessage call may match mapping C instead of mapping B.

So replay returns:
Record used: B1
Replay uses: C1 (from mapping C recorded earlier)

Both B1 and C1 were generated by server side. replay is picking a different recorded mapping because the request no longer matches the same one.

impl sendAck and sendNack

dddc18e

salesforce-cla bot added the cla:signed label Nov 7, 2025

sandeepvinayak reviewed Nov 10, 2025

View reviewed changes

roseyang62 added 8 commits November 11, 2025 13:04

address comment

8c9c960

address comments

cc574fb

refactored and removed AckID interface

c7b0c4e

remove transformer and record again

8f299dc

fix replay

50e010d

add updated transfomer

cba142f

add javadoc

b47cd34

update javadoc

e927a09

pubsub: Enable sendAck and sendNack apis for AWS Pubsub #134

Are you sure you want to change the base?

pubsub: Enable sendAck and sendNack apis for AWS Pubsub #134

Uh oh!

Conversation

roseyang62 commented Nov 7, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Some conventions to follow

Uh oh!

codecov-commenter commented Nov 7, 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!

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!

sandeepvinayak Nov 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

roseyang62 commented Nov 7, 2025 •

edited

Loading

codecov-commenter commented Nov 7, 2025 •

edited

Loading

sandeepvinayak Nov 14, 2025 •

edited

Loading