hudi-agent commented on code in PR #19396:
URL: https://github.com/apache/hudi/pull/19396#discussion_r3686693236
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/config/S3SourceConfig.java:
##########
@@ -90,4 +96,18 @@ public class S3SourceConfig extends HoodieConfig {
.markAdvanced()
.withDocumentation("Visibility timeout for messages in queue. After we
consume the message, queue will move the consumed "
+ "messages to in-flight state, these messages can't be consumed
again by source for this timeout period.");
+
+ public static final ConfigProperty<Integer>
S3_SOURCE_QUEUE_PROCESSING_PARALLELISM = ConfigProperty
+ .key(S3_SOURCE_PREFIX + "queue.processing.parallelism")
+ .defaultValue(16)
Review Comment:
🤖 The default here is 16, so every existing `S3EventsSource` deployment
flips from serial to 16-way concurrent SQS calls on upgrade rather than opting
in. The design does mitigate this well — worker count is `min(parallelism,
plannedReceiveCalls)` so idle/small queues stay effectively serial, and the
SDK's default pool of 50 covers 16 — but was a non-1 default intended for all
existing users here, or would defaulting to 1 (preserve today's behaviour, let
operators opt into the speedup) be the safer rollout? @yihua does the ingestion
behaviour change on upgrade seem acceptable to you?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
##########
hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/helpers/CloudObjectsSelector.java:
##########
@@ -197,48 +789,452 @@ protected List<List<MessageTracker>>
createListPartitions(List<MessageTracker> s
}
/**
- * Delete batch of messages from queue.
+ * Deletes one batch (at most {@value #SQS_BATCH_MAX_ENTRIES} entries, the
SQS API cap) of messages
+ * and returns the trackers SQS reported as failed, so the caller can retry
them. DeleteMessageBatch
+ * can return partial failures even on an HTTP 200, and undeleted messages
stay in-flight and are
+ * redelivered once the visibility timeout expires, so failures must be
surfaced rather than only
+ * counted.
+ *
+ * <p>A synthetic per-entry id (the batch-local index) is used instead of
the message id: the id
+ * only has to be unique within the batch, and under at-least-once delivery
the same message id can
+ * appear twice, which would otherwise collide and make SQS reject the whole
batch.
*/
- protected void deleteBatchOfMessages(SqsClient sqs, String queueUrl,
List<MessageTracker> messagesToBeDeleted) {
+ protected List<FailedDelete> deleteBatchOfMessages(SqsClient sqs, String
queueUrl, List<MessageTracker> messagesToBeDeleted) {
if (messagesToBeDeleted.isEmpty()) {
- return;
+ return Collections.emptyList();
}
- DeleteMessageBatchRequest.Builder builder =
DeleteMessageBatchRequest.builder().queueUrl(queueUrl);
- List<DeleteMessageBatchRequestEntry> deleteEntries = new ArrayList<>();
-
- for (MessageTracker message : messagesToBeDeleted) {
+ ValidationUtils.checkArgument(messagesToBeDeleted.size() <=
SQS_BATCH_MAX_ENTRIES,
+ "DeleteMessageBatch accepts at most " + SQS_BATCH_MAX_ENTRIES + "
entries per call, got "
+ + messagesToBeDeleted.size());
+ List<DeleteMessageBatchRequestEntry> deleteEntries = new
ArrayList<>(messagesToBeDeleted.size());
+ for (int i = 0; i < messagesToBeDeleted.size(); i++) {
deleteEntries.add(
DeleteMessageBatchRequestEntry.builder()
- .id(message.messageId)
- .receiptHandle(message.receiptHandle)
+ .id(String.valueOf(i))
+ .receiptHandle(messagesToBeDeleted.get(i).receiptHandle)
.build());
}
- builder.entries(deleteEntries);
- DeleteMessageBatchResponse deleteResponse =
sqs.deleteMessageBatch(builder.build());
- List<String> deleteFailures =
- deleteResponse.failed().stream()
- .map(BatchResultErrorEntry::id)
- .collect(Collectors.toList());
- if (!deleteFailures.isEmpty()) {
- log.warn(
- "Failed to delete {} messages out of {} from queue.",
deleteFailures.size(), deleteEntries.size());
- } else {
+ DeleteMessageBatchResponse deleteResponse = sqs.deleteMessageBatch(
+
DeleteMessageBatchRequest.builder().queueUrl(queueUrl).entries(deleteEntries).build());
+ if (deleteResponse.failed().isEmpty()) {
log.debug("Successfully deleted {} messages from queue.",
deleteEntries.size());
+ return Collections.emptyList();
+ }
+ // Keep the SQS-provided reason (code + senderFault) with each failed
tracker so the caller can log
+ // why deletion failed once at the end rather than per batch.
+ List<FailedDelete> failed = new
ArrayList<>(deleteResponse.failed().size());
+ for (BatchResultErrorEntry error : deleteResponse.failed()) {
+ MessageTracker tracker = trackerForEntryId(messagesToBeDeleted,
error.id());
+ if (tracker == null) {
+ // SQS echoed an entry id that was never sent (ids are batch-local
indices assigned just above, so
+ // this is a protocol violation rather than something that happens in
practice). The failure cannot
+ // be attributed to a message: there is no receipt handle to retry
with, and it cannot be added to
+ // the residual set either, since that set is counted against
processedMessages to derive the
+ // deleted total and a phantom entry would corrupt it. So it is
reported here and nowhere else -
+ // meaning the summary's deleted count is an upper bound whenever this
WARN fires. The affected
+ // message stays in-flight and is redelivered after the visibility
timeout regardless.
+ log.warn("SQS reported a delete failure for unknown entry id \"{}\" on
queue {} (code={}); it cannot be "
+ + "mapped back to a message, so that message's deletion status is
unknown and the deleted count "
+ + "below may over-count by one.",
+ error.id(), queueUrl, error.code());
+ continue;
+ }
+ failed.add(FailedDelete.reported(tracker, error));
+ }
+ // Per-batch detail at DEBUG only; residual failures are summarized once
at WARN in
+ // deleteProcessedMessages so a transient blip across many batches does
not flood the logs.
+ log.debug("Failed to delete {} messages out of {} from queue {}.",
failed.size(), deleteEntries.size(), queueUrl);
+ return failed;
+ }
+
+ /**
+ * Resolves an entry id assigned by {@link #deleteBatchOfMessages} back to
its message. The id is the
+ * batch-local index, so no lookup map is needed. Returns {@code null} when
SQS echoes an id that was
+ * never sent (non-numeric or out of range), which the caller reports rather
than ignores.
+ */
+ private static MessageTracker trackerForEntryId(List<MessageTracker> batch,
String entryId) {
+ try {
+ int index = Integer.parseInt(entryId);
+ return index >= 0 && index < batch.size() ? batch.get(index) : null;
+ } catch (NumberFormatException e) {
+ return null;
}
}
/**
* Delete Queue Messages after hudi commit. This method will be invoked by
source.onCommit.
+ * Deletes run concurrently across a fixed pool of up to {@link
#processingParallelism} threads;
+ * batches that fail - whether SQS reported the entry as failed on an HTTP
200 or the call threw
+ * outright - are collected across all batches and retried up to {@value
#DELETE_MAX_RETRIES} times
+ * with exponential backoff, since undeleted messages stay in-flight,
consume the in-flight quota, and
+ * are redelivered once the visibility timeout expires. Any residual
failures after the retries are
+ * logged with their reason (see {@link #logDeleteFailures}).
+ *
+ * <p>A failed delete is deliberately not fatal. This runs from {@code
Source.onCommit}, which
+ * StreamSync calls <em>after</em> the Hudi commit has already landed, and
HoodieStreamer turns any
+ * exception out of the sync round into a job shutdown - so aborting here on
one throttled call would
+ * take ingestion down over a condition SQS itself recovers from by
redelivering. The one exception is
+ * a systemic failure: if calls were throwing and not a single message could
be deleted, that is a
+ * queue/credential/network problem rather than stale individual entries,
and it is surfaced.
*/
public void deleteProcessedMessages(SqsClient sqs, String queueUrl,
List<MessageTracker> processedMessages) {
- if (!processedMessages.isEmpty()) {
- // create batch for deletion, SES DeleteMessageBatchRequest only accept
max 10 entries
- List<List<MessageTracker>> deleteBatches =
createListPartitions(processedMessages, 10);
- for (List<MessageTracker> deleteBatch : deleteBatches) {
- deleteBatchOfMessages(sqs, queueUrl, deleteBatch);
+ if (processedMessages.isEmpty()) {
+ return;
+ }
+ long startMs = System.currentTimeMillis();
+ // create batch for deletion, DeleteMessageBatchRequest only accepts max
SQS_BATCH_MAX_ENTRIES entries
+ List<List<MessageTracker>> deleteBatches =
createListPartitions(processedMessages, SQS_BATCH_MAX_ENTRIES);
+ int totalBatches = deleteBatches.size();
+ int workers = Math.max(1, Math.min(processingParallelism, totalBatches));
+ AtomicLong busyNanos = new AtomicLong();
+ ExecutorService pool = newFixedThreadPool(DELETE_THREAD_PREFIX, workers);
+ // Failures SQS attributes to this caller (senderFault=true, e.g.
ReceiptHandleIsInvalid once the
+ // visibility timeout has expired) cannot succeed on a retry, so they are
set aside by the pass that
+ // reported them instead of consuming the retry budget - and are still
reported at the end.
+ List<FailedDelete> permanentFailures = new ArrayList<>();
+ DeleteStats stats = new DeleteStats();
+ int retries = 0;
+ try {
+ DeletePass pass = deleteBatchesConcurrently(pool, sqs, queueUrl,
deleteBatches, workers, busyNanos);
+ stats.add(pass);
+ List<FailedDelete> retryable = splitOffPermanentFailures(pass,
permanentFailures);
+ while (!retryable.isEmpty() && retries < DELETE_MAX_RETRIES) {
+ retries++;
+ long backoffMs = sleepBeforeDeleteRetry(queueUrl, retryable.size(),
retries);
+ List<MessageTracker> retryTrackers =
+ retryable.stream().map(failedDelete ->
failedDelete.message).collect(Collectors.toList());
+ pass = deleteBatchesConcurrently(pool, sqs, queueUrl,
+ createListPartitions(retryTrackers, SQS_BATCH_MAX_ENTRIES),
workers, busyNanos);
+ stats.add(pass);
+ retryable = splitOffPermanentFailures(pass, permanentFailures);
+ log.debug("Delete retry {} for queue {} (backoff {} ms) left {}
messages still failing.",
+ retries, queueUrl, backoffMs, retryable.size());
+ }
+ List<FailedDelete> residual = new ArrayList<>(permanentFailures);
+ residual.addAll(retryable);
+ long wallMs = System.currentTimeMillis() - startMs;
+ int deleted = processedMessages.size() - residual.size();
+ log.info("Deleted {} of {} processed messages from queue {} across {}
delete batches and {} SQS calls "
+ + "({} calls failed, {} retry passes) in {} ms ({} messages
failed to delete).",
+ deleted, processedMessages.size(), queueUrl, totalBatches,
stats.calls, stats.failedCalls,
+ retries, wallMs, residual.size());
+ if (stats.failedCalls > 0) {
+ // The residual WARN below groups by reason but carries no stack
trace; log the first cause once
+ // here so the SDK-level reason (throttling, connection acquisition
timeout, ...) is recoverable.
+ log.warn("{} of {} SQS DeleteMessageBatch calls failed for queue {};
their batches were retried and "
+ + "{} messages remain undeleted. Undeleted messages are
redelivered after the visibility "
+ + "timeout, so ingestion may see them again rather than losing
them.",
+ stats.failedCalls, stats.calls, queueUrl, residual.size(),
stats.firstThrown);
+ }
+ if (stats.fatalThrown != null) {
+ // A non-retryable failure that struck only part of the phase (a
session token expiring mid-run,
+ // say) leaves deleted > 0, so the systemic throw below stays silent
by design. Without this it
+ // would be reported only as an indistinguishable residual-failure
WARN, so name the condition
+ // explicitly: no retry can clear it, and it will recur on every
commit until an operator acts.
+ log.error("Non-retryable SQS DeleteMessageBatch failure for queue {};
stopped dispatching further "
+ + "delete batches ({} batches skipped) and did not retry,
since no retry can succeed. {} of {} "
+ + "messages remain undeleted and will be redelivered after the
visibility timeout. Check the "
+ + "queue url, its region, and the credentials/permissions of
this job.",
+ queueUrl, stats.skippedBatches, residual.size(),
processedMessages.size(), stats.fatalThrown);
+ }
+ if (!residual.isEmpty()) {
+ logDeleteFailures(queueUrl, residual, retries);
+ }
+ logParallelPerf("delete", queueUrl, workers, stats.calls, wallMs,
busyNanos.get());
+ // Nothing at all could be deleted and calls were throwing: systemic
(bad credentials, queue gone,
+ // network down) rather than individual stale entries, and the caller is
about to clear its tracked
+ // messages as if they had been handled, so surface it. A purely
senderFault residue is deliberately
+ // not fatal - every receipt handle being stale is a real queue state
that no retry or job failure
+ // can fix, and the messages are redelivered regardless.
+ //
+ // NOTE: this deliberately does not fire when every entry failed
server-side on an HTTP 200
+ // (senderFault=false, e.g. InternalError) with no call ever throwing.
That is an equally total
+ // failure to delete, but SQS reports genuinely broken plumbing - bad
credentials, a deleted queue -
+ // by throwing, so firstThrown is the sharper signal, and escalating a
persistent per-entry
+ // condition into a job shutdown would trade at-least-once redelivery
for an ingestion outage.
+ // Making that case visible belongs to the residual-failure metric
tracked as follow-up, not here.
+ // The extra answeredCalls() guard: a DeleteMessageBatch that returned
HTTP 200 proves the queue is
+ // reachable and the credentials work, even if it reported every entry
as failed. Without it, the
+ // deliberately-non-fatal all-senderFault case above (every receipt
handle stale after a long commit)
+ // turns into a job shutdown as soon as one unrelated call is also
throttled - and fanning out makes
+ // such a throttle more likely, not less.
+ if (deleted == 0 && stats.firstThrown != null && stats.answeredCalls()
== 0) {
+ throw new HoodieException("Failed to delete any of the " +
processedMessages.size()
+ + " processed messages from SQS queue " + queueUrl + " (" +
stats.failedCalls + " of "
+ + stats.calls + " DeleteMessageBatch calls threw, " +
residual.size()
+ + " messages still undeleted after " + retries + " retries)",
stats.firstThrown);
+ }
+ } finally {
+ shutdownThreadPool(pool);
+ }
+ }
+
+ /**
+ * Deletes the given batches concurrently on {@code pool} and returns the
outcome of the pass: every
+ * message that was not deleted, plus the call stats. Each batch's SQS call
time is accumulated into
+ * {@code busyNanos} so the caller can log the serialized-equivalent time
and prove the deletes
+ * actually overlapped.
+ *
+ * @param workers the pool's size, passed in rather than re-derived from
{@code deleteBatches}. A retry
+ * pass carries fewer batches than the first one, so deriving it here
would produce a second,
+ * smaller value under the same name as the caller's - and the caller's
is the one that sized the
+ * pool and is reported as {@code parallelism} on the perf line.
Over-dispatch is not a concern:
+ * {@code pending} is exhausted once every batch is submitted, whatever
the slot count says.
+ */
+ private DeletePass deleteBatchesConcurrently(ExecutorService pool, SqsClient
sqs, String queueUrl,
+ List<List<MessageTracker>>
deleteBatches, int workers,
+ AtomicLong busyNanos) {
+ CompletionService<BatchOutcome> completionService = new
ExecutorCompletionService<>(pool);
+ Iterator<List<MessageTracker>> pending = deleteBatches.iterator();
+ DeletePass pass = new DeletePass();
+ int outstanding = 0;
+ try {
+ // Dispatch incrementally rather than queueing every batch up front. The
pool stays saturated either
+ // way, but holding the untried batches back is what makes fail-fast
possible: on a non-retryable
+ // failure the remaining batches are simply never submitted. With a
deleted queue or revoked
+ // credentials that is the difference between ceil(M/10) doomed calls
and roughly one per worker.
Review Comment:
🤖 nit: could you use an explicit `case TRANSIENT:` instead of `default:`
here? With `default` a future `SqsFailureKind` value would silently be treated
as transient with no compile-time signal that a new case was missed.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag
quality.</i></sub>
--
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]
