This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new f8e9a768b634 CAMEL-25822: camel-aws2-sqs - the DST world drains until
no progress instead of for a fixed time (#28139)
f8e9a768b634 is described below
commit f8e9a768b634e88a75b831bcd20df50a3e64624c
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 21 13:56:30 2026 +0530
CAMEL-25822: camel-aws2-sqs - the DST world drains until no progress
instead of for a fixed time (#28139)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/test-dst.adoc | 16 ++++++++++++++--
.../apache/camel/component/aws2/sqs/dst/SqsWorld.java | 15 +++++++++++++--
.../camel-test-dst/src/main/docs/test-dst.adoc | 16 ++++++++++++++--
3 files changed, 41 insertions(+), 6 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
index 7a001ec87504..74a55ce94812 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
@@ -226,12 +226,24 @@ dst.runFor(dst.between(0, 3L * visibilityTimeout * 1000));
// faults cleared: the queue must drain
faults = false;
sqs.faults = false;
-long deadline = dst.now() + 20L * visibilityTimeout * 1000 + 30_000;
-while (!sqs.allDeleted() && dst.now() < deadline) {
+// drain while it makes progress: stop when no message was deleted for a
while, or at a cap
+long quiet = 20L * visibilityTimeout * 1000 + 30_000;
+long cap = dst.now() + 3_600_000;
+int left = sqs.notDeleted().size();
+long lastProgress = dst.now();
+while (!sqs.allDeleted() && dst.now() < cap && dst.now() - lastProgress <
quiet) {
dst.runFor(100);
+ int remaining = sqs.notDeleted().size();
+ if (remaining < left) {
+ left = remaining;
+ lastProgress = dst.now();
+ }
}
----
+Drain until progress stops, not for a fixed time. With a fixed drain time, 1
of 20,000 seeds failed falsely: one
+consumer taking one message per poll was still deleting messages when the time
ran out.
+
Some faults legitimately cause what the invariant forbids. Record which fault
excuses which violation, and only those:
a failed `ChangeMessageVisibilityBatch` lets SQS deliver the messages of that
batch again, so `FakeSqs` reports their
ids (`onExtensionFault`) and the world does not count a redelivery of those
messages.
diff --git
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
index e8096e53480a..56873b7dc080 100644
---
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
+++
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/dst/SqsWorld.java
@@ -112,9 +112,20 @@ final class SqsWorld {
// faults cleared: the queue must drain
faults = false;
sqs.faults = false;
- long deadline = dst.now() + 20L * visibilityTimeout * 1000 +
30_000;
- while (!sqs.allDeleted() && dst.now() < deadline) {
+ // drain while it makes progress: stop when no message was deleted
for a while, or at a cap. A fixed
+ // drain time failed falsely when the consumer was still deleting
messages at its end (one consumer
+ // taking one message per poll with a 1 s delay, and short polling
returning fewer messages than asked)
+ long quiet = 20L * visibilityTimeout * 1000 + 30_000;
+ long cap = dst.now() + 3_600_000;
+ int left = sqs.notDeleted().size();
+ long lastProgress = dst.now();
+ while (!sqs.allDeleted() && dst.now() < cap && dst.now() -
lastProgress < quiet) {
dst.runFor(100);
+ int remaining = sqs.notDeleted().size();
+ if (remaining < left) {
+ left = remaining;
+ lastProgress = dst.now();
+ }
}
dst.check(
Invariants.of("no message is delivered again while the
consumer holds it",
diff --git a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
index 7a001ec87504..74a55ce94812 100644
--- a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
+++ b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
@@ -226,12 +226,24 @@ dst.runFor(dst.between(0, 3L * visibilityTimeout * 1000));
// faults cleared: the queue must drain
faults = false;
sqs.faults = false;
-long deadline = dst.now() + 20L * visibilityTimeout * 1000 + 30_000;
-while (!sqs.allDeleted() && dst.now() < deadline) {
+// drain while it makes progress: stop when no message was deleted for a
while, or at a cap
+long quiet = 20L * visibilityTimeout * 1000 + 30_000;
+long cap = dst.now() + 3_600_000;
+int left = sqs.notDeleted().size();
+long lastProgress = dst.now();
+while (!sqs.allDeleted() && dst.now() < cap && dst.now() - lastProgress <
quiet) {
dst.runFor(100);
+ int remaining = sqs.notDeleted().size();
+ if (remaining < left) {
+ left = remaining;
+ lastProgress = dst.now();
+ }
}
----
+Drain until progress stops, not for a fixed time. With a fixed drain time, 1
of 20,000 seeds failed falsely: one
+consumer taking one message per poll was still deleting messages when the time
ran out.
+
Some faults legitimately cause what the invariant forbids. Record which fault
excuses which violation, and only those:
a failed `ChangeMessageVisibilityBatch` lets SQS deliver the messages of that
batch again, so `FakeSqs` reports their
ids (`onExtensionFault`) and the world does not count a redelivery of those
messages.
