Davis-Zhang-Onehouse opened a new pull request, #19396: URL: https://github.com/apache/hudi/pull/19396
## What `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. This fans both phases out across a shared bounded thread pool. ## Continuous dispatch rather than waves Calls are dispatched through an `ExecutorCompletionService`: the coordinating thread consumes each completion as it arrives and immediately refills the freed slot. The obvious alternative — `invokeAll` over a batch of calls in a loop — is a barrier, so each 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, so nothing is shared with the workers beyond a timing accumulator. Admission credits each in-flight call with a full `maxMessagesPerRequest`, so a call is never dispatched that the messages already in flight would make redundant. That bounds overshoot of `maxMessagePerBatch` to `maxMessagesPerRequest - 1`. ## Drain detection scaled to poll cost The previous loop ended the entire batch on the first empty response. AWS documents that as unreliable: > In rare cases, you might receive empty responses even when a queue still contains messages, especially if you specified a low value for the `WaitTimeSeconds` parameter. It is also the expensive case. Under long polling: > ReceiveMessage queries all servers for messages, sending a response once at least one message is available, up to the specified maximum. An empty response is sent only if the polling wait time expires. So an empty response costs a full `WaitTimeSeconds` (20 s at the maximum), while a non-empty one returns as soon as a message is available. Requiring however many empty responses fit in one ~20 s window bounds drain confirmation to roughly a single poll window at any setting, with no timer and nothing to abort mid-flight: | `WaitTimeSeconds` | empty responses required | |---|---| | 20 (AWS maximum) | 1 | | 10 | 2 | | 5 | 4 | | 1 | 20 | | 0 (short poll) | 5 | Short polling samples only a subset of servers, so its empty responses are weak evidence but nearly free — hence the higher fixed count there. Empty responses are counted as a batch total rather than consecutively, since resetting the count on a stray message lets a queue with slow ingress hold the phase for its entire call budget. Confirming a drain early never drops messages; whatever is not received stays in the queue for the next batch. ## Failure classification SQS status codes do not track severity — `RequestThrottled` and `OverLimit` are both HTTP 400, `ThrottlingException` is 403, `MalformedQueryString` is 404 — so failures are classified by error code, with the HTTP status used only as a fallback for codes that are not recognised (an unknown 5xx is transient, an unknown 4xx is not). `AwsErrorCode.THROTTLING_ERROR_CODES` omits `KmsThrottled`, which is therefore covered explicitly. `OverLimit` (the in-flight message ceiling) is treated as backpressure rather than 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 quota. Failing the job over normal saturation would be worse than pausing. Non-retryable failures stop at the first occurrence instead of consuming the retry budget on a condition no retry can fix. Calls already 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. For the same reason an empty result only surfaces as an error when nothing at all proved the queue reachable; a successful response, or an `OverLimit` returned by SQS itself, is evidence the endpoint, credentials and queue are healthy. ## Delete Partial failures are aggregated across all batches and retried with exponential backoff, since undeleted messages stay in flight, consume the in-flight quota, and are redelivered once the visibility timeout expires. Entries SQS marks `senderFault` are set aside rather than retried, as no retry can fix them. Failures that a retry genuinely cannot help — a deleted queue, revoked credentials — stop dispatch instead of being retried across every remaining batch. 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. ## Config New: `hoodie.streamer.s3.source.queue.processing.parallelism` (default 16). Set to 1 for the previous fully sequential behaviour. The shared `SqsClient` is thread-safe and intended to be shared. 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. ## Testing `TestCloudObjectsSelector` 50/50, plus `TestS3EventsMetaSelector` and `TestS3EventsSource`. Covers concurrent receive and delete (every message fetched exactly once across 8 workers), the drain-count table, a spurious empty response not truncating a batch, overshoot bounded by in-flight accounting, each failure class end to end (transient tolerated, non-retryable failing fast, `OverLimit` returning cleanly), the classifier itself, and `Error` propagation. Two tests are regression detectors rather than plain assertions: - one holds a single call until its siblings have completed several further calls between them, so a return to wave-based dispatch fails the test rather than passing quietly; - one pins the deliberate short-poll trade-off, so a change to how empty responses are counted cannot alter it silently. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
