perf(ingest): parallelize SQS receive and delete in the S3 events source - #19396
Open
Davis-Zhang-Onehouse wants to merge 1 commit into
Open
perf(ingest): parallelize SQS receive and delete in the S3 events source#19396Davis-Zhang-Onehouse wants to merge 1 commit into
Davis-Zhang-Onehouse wants to merge 1 commit into
Conversation
S3EventsSource drains its SQS queue serially, 10 messages at a time, for
both receive and delete. ReceiveMessage and DeleteMessageBatch are capped
by SQS at 10 entries per call, so draining a backlog is dominated by
round-trip latency and concurrency is the only available lever. Both
phases now fan out across a shared bounded thread pool.
Calls are dispatched through an ExecutorCompletionService rather than in
waves: the coordinating thread consumes each completion as it arrives and
immediately refills the freed slot. invokeAll over a batch of calls in a
loop is a barrier, so a wave costs max(latency) rather than avg(latency),
and no worker starts its next call until the slowest sibling returns. All
counters and termination state stay on the coordinating thread.
The previous loop ended the whole batch on the first empty response, which
AWS documents as unreliable ("in rare cases, you might receive empty
responses even when a queue still contains messages"). It is also the
expensive case: under long polling "an empty response is sent only if the
polling wait time expires", so an empty costs a full WaitTimeSeconds while
a non-empty one returns as soon as a message is available. Requiring
however many empty responses fit in one ~20s window bounds drain
confirmation to roughly a single poll window at any setting, with no timer
- 1 empty at WaitTimeSeconds=20, 2 at 10, 4 at 5, 20 at 1. Short polling
samples only a subset of servers, so its empties are weak evidence but
nearly free, and keep a higher fixed count.
Failures are classified by SQS error code, with HTTP status only as the
fallback for unrecognised codes, because SQS status codes do not track
severity: RequestThrottled and OverLimit are both 400, ThrottlingException
is 403, MalformedQueryString is 404. AwsErrorCode.THROTTLING_ERROR_CODES
omits KmsThrottled, so that is covered explicitly. OverLimit is the
in-flight ceiling and is treated as backpressure rather than an error,
since the received messages become deletable once the batch commits, which
is what frees the quota. Non-retryable failures stop at the first
occurrence instead of consuming the retry budget.
Calls in flight when dispatch stops are drained and their messages kept,
never cancelled: a ReceiveMessage that succeeded server-side has already
made those messages invisible, so discarding the response would strand
them for the full visibility timeout. An empty result surfaces as an error
only when nothing at all proved the queue reachable.
DeleteMessageBatch partial failures are aggregated across batches and
retried with exponential backoff; senderFault entries are set aside, as no
retry can fix them. A synthetic per-entry id (the batch-local index)
replaces the message id so duplicate ids within a batch under at-least-once
delivery cannot collide and cause SQS to reject the whole batch. Deletes
remain in Source.onCommit.
Adds hoodie.streamer.s3.source.queue.processing.parallelism (default 16;
set to 1 to restore fully sequential behaviour). The SDK's default sync
connection pool of 50 already covers any value up to 50, so only a larger
value causes the source to resize the pool.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19396 +/- ##
============================================
+ Coverage 72.49% 72.55% +0.05%
- Complexity 32888 32979 +91
============================================
Files 2574 2574
Lines 149010 149383 +373
Branches 18749 18807 +58
============================================
+ Hits 108032 108386 +354
- Misses 32513 32528 +15
- Partials 8465 8469 +4
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
Collaborator
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe the issue this Pull Request addresses
S3EventsSourcedrains its SQS queue serially, 10 messages at a time, for both receive and delete.ReceiveMessageandDeleteMessageBatchare capped by SQS at 10 entries per call, so draining a backlog is dominated by round-trip latency and concurrency is the only available lever. On a deep backlog these two serial phases dominate the ingestion cycle.Two correctness problems in the existing loop compound this:
WaitTimeSecondsparameter."DeleteMessageBatchpartial failures are counted but never retried, and undeleted messages stay in flight, consume the in-flight quota, and are redelivered once the visibility timeout expires.Summary and Changelog
Users draining a large SQS backlog through
S3EventsSourceget a substantially shorter fetch/commit cycle, since the receive and delete round-trips now overlap instead of running one at a time. Behaviour is unchanged atparallelism=1.Changes:
ExecutorCompletionService: the coordinating thread consumes each completion as it arrives and immediately refills the freed slot.invokeAllover a batch of calls in a loop was rejected as it is a barrier — a wave costsmax(latency)rather thanavg(latency), and no worker starts its next call until the slowest sibling returns. All counters and termination state stay on the coordinating thread; nothing is shared with the workers beyond a timing accumulator.maxMessagesPerRequest, so a call is never dispatched that messages already in flight would make redundant. Overshoot ofmaxMessagePerBatchis bounded tomaxMessagesPerRequest - 1.WaitTimeSecondswhile a non-empty one returns as soon as a message is available. The batch now requires however many empty responses fit in one ~20s window — 1 atWaitTimeSeconds=20, 2 at 10, 4 at 5, 20 at 1 — which bounds drain confirmation to roughly a single poll window at any setting, with no timer. Short polling samples only a subset of servers, so its empty responses are weak evidence but nearly free, and keep a higher fixed count (5). Empty responses are counted as a batch total rather than consecutively, since resetting on a stray message lets a queue with slow ingress hold the phase for its entire call budget.RequestThrottledandOverLimitare both HTTP 400,ThrottlingExceptionis 403,MalformedQueryStringis 404.AwsErrorCode.THROTTLING_ERROR_CODESomitsKmsThrottled, which is covered explicitly.OverLimittreated as backpressure, not an error: it stops dispatch and returns what was fetched, since the received messages become deletable once the batch commits, which is what frees the in-flight quota. Non-retryable failures stop at the first occurrence rather than consuming the retry budget.ReceiveMessagethat succeeded server-side has already made those messages invisible, so discarding the response would strand them for the full visibility timeout. Correspondingly, an empty result only surfaces as an error when nothing at all proved the queue reachable — a successful response, or anOverLimitreturned by SQS itself, is evidence the endpoint, credentials and queue are healthy.senderFaultentries are set aside rather than retried, as no retry can fix them. A synthetic per-entry id (the batch-local index) replaces the message id, so duplicate message ids within a batch under at-least-once delivery cannot collide and cause SQS to reject the whole batch; the id doubles as the reverse lookup, so no map is needed.Deletes remain in
Source.onCommit. Deleting inside the receive loop would lift the in-flight ceiling but would break the coupling between message deletion and the commit that consumed those messages.No code was copied from another project.
Impact
hoodie.streamer.s3.source.queue.processing.parallelism, default 16. Setting it to 1 restores the previous fully sequential behaviour.SqsClientis thread-safe and remains a single instance. Its HTTP connection pool must be at least the worker count or workers block on connection acquisition; the SDK's default sync pool (maxConnections=50) already covers any value up to 50, so only a larger value causes the source to resize the pool — which is whysoftware.amazon.awssdk:apache-clientis declaredprovidedrather than made a hard runtime requirement.Risk Level
Medium. The change makes a previously single-threaded ingestion path concurrent and alters when a receive batch terminates.
Mitigations and verification:
TestCloudObjectsSelector50/50, plusTestS3EventsMetaSelectorandTestS3EventsSource; fullmvn clean installclean.OverLimitreturning cleanly with and without fetched messages — plus the classifier itself as a unit table, andErrorpropagation.parallelism=1is retained as an escape hatch to the previous sequential behaviour.Documentation Update
The new config
hoodie.streamer.s3.source.queue.processing.parallelismcarries a fullwithDocumentationdescription covering the SQS 10-per-call cap, the connection-pool requirement, and theparallelism=1escape hatch.hoodie.streamer.s3.source.queue.long.poll.waithas its description extended, since it now also governs how many empty responses confirm a drain.No website changes are needed — both are existing/advanced source configs surfaced through the generated config docs.
Contributor's checklist