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 d09593b4a584 CAMEL-25700: camel-aws2-sqs - extend the visibility of
every message of the batch, not only from its turn (#27985)
d09593b4a584 is described below
commit d09593b4a5847fb24ee47fa0b12dace2890e3762
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 17 16:06:00 2026 +0530
CAMEL-25700: camel-aws2-sqs - extend the visibility of every message of the
batch, not only from its turn (#27985)
With extendMessageVisibility=true and maxMessagesPerPoll greater than one,
processBatch registered a
message with the TimeoutExtender only when its turn came. The visibility
timeout of all messages of a
poll starts at the receive, so the messages that waited behind slow ones
were not extended at all and
became visible again: SQS delivered them a second time (the scenario of
CAMEL-10587).
processBatch now registers every message of the batch with the extender
before it routes the first
one, and removes the messages that it does not route (the consumer is
stopping or suspending, or the
processor threw) so that they become visible again as before. The
extender's rule of extending an
entry only when its deadline is near is unchanged.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/aws2/sqs/Sqs2Consumer.java | 45 +++++--
...qsConsumerExtendMessageVisibilityBatchTest.java | 147 +++++++++++++++++++++
2 files changed, 184 insertions(+), 8 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
index e51d2912a363..99cb666cd3ae 100644
---
a/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
+++
b/components/camel-aws/camel-aws2-sqs/src/main/java/org/apache/camel/component/aws2/sqs/Sqs2Consumer.java
@@ -136,6 +136,16 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
public int processBatch(Queue<Object> exchanges) throws Exception {
int total = exchanges.size();
+ // the visibility timeout of every message of the batch runs from the
receive, so all of them are extended from
+ // now on, and not only when their turn comes: otherwise a message
that waits behind slow ones becomes visible
+ // again and is delivered a second time
+ final TimeoutExtender extender = this.timeoutExtender;
+ if (extender != null) {
+ for (Object exchange : exchanges) {
+ extender.add((Exchange) exchange);
+ }
+ }
+
for (int index = 0; index < total && isBatchAllowed(); index++) {
// only loop if we are started (allowed to run)
final Exchange exchange = ObjectHelper.cast(Exchange.class,
exchanges.poll());
@@ -147,10 +157,6 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
// update pending number of exchanges
pendingExchanges = total - index - 1;
- if (this.timeoutExtender != null) {
- timeoutExtender.add(exchange);
- }
-
// add on completion to handle after work when the exchange is done
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
@Override
@@ -174,18 +180,33 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
try {
getAsyncProcessor().process(exchange, cb);
} catch (Error e) {
- if (this.timeoutExtender != null) {
+ if (extender != null) {
// fatal error so stop the timeout extender
- timeoutExtender.cancel();
- timeoutExtender.entries.clear();
+ extender.cancel();
+ extender.entries.clear();
}
throw e;
+ } catch (RuntimeException e) {
+ stopExtending(extender, exchanges);
+ throw e;
}
}
+ // the messages that are not routed (the consumer is stopping or
suspending) are no longer extended, so they
+ // become visible again when their current visibility timeout runs out
+ stopExtending(extender, exchanges);
+
return total;
}
+ private static void stopExtending(TimeoutExtender extender, Queue<Object>
notRouted) {
+ if (extender != null) {
+ for (Object exchange : notRouted) {
+ extender.remove((Exchange) exchange);
+ }
+ }
+ }
+
/**
* Strategy to delete the message after being processed.
*
@@ -412,7 +433,7 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
private void remove(Exchange exchange) {
LOG.trace("Removing exchangeId {} from the
TimeoutExtender, processing done", exchange.getExchangeId());
- entries.remove(exchange.getExchangeId());
+ TimeoutExtender.this.remove(exchange);
}
});
@@ -423,6 +444,14 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
entries.put(exchange.getExchangeId(), new
TimeoutExtenderEntry(entry));
}
+ /**
+ * Stops extending the visibility of the message of the given
exchange. The message is not made visible: it
+ * becomes visible again when its current (possibly just extended)
visibility timeout runs out.
+ */
+ public void remove(Exchange exchange) {
+ entries.remove(exchange.getExchangeId());
+ }
+
public void cancel() {
// cancel by setting to no longer run
run.set(false);
diff --git
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsConsumerExtendMessageVisibilityBatchTest.java
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsConsumerExtendMessageVisibilityBatchTest.java
new file mode 100644
index 000000000000..39de83b93b2a
--- /dev/null
+++
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsConsumerExtendMessageVisibilityBatchTest.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.aws2.sqs;
+
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.EndpointInject;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import
software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchRequest;
+import
software.amazon.awssdk.services.sqs.model.ChangeMessageVisibilityBatchResponse;
+import software.amazon.awssdk.services.sqs.model.Message;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * With {@code extendMessageVisibility=true} and more than one message per
poll, the messages that wait in the batch for
+ * their turn must be kept invisible too: their visibility timeout runs from
the receive, not from the start of their
+ * own processing.
+ */
+class SqsConsumerExtendMessageVisibilityBatchTest extends CamelTestSupport {
+
+ private static final int VISIBILITY_TIMEOUT_SECONDS = 2;
+ private static final String FIRST_HANDLE = "receipt-handle-1";
+ private static final String SECOND_HANDLE = "receipt-handle-2";
+
+ private final CountDownLatch firstStarted = new CountDownLatch(1);
+ private final CountDownLatch releaseFirst = new CountDownLatch(1);
+ private final CountDownLatch secondExtended = new CountDownLatch(1);
+ private final ExecutorService suspender =
Executors.newSingleThreadExecutor();
+
+ @EndpointInject("mock:result")
+ private MockEndpoint mock;
+
+ @BindToRegistry("amazonSQSClient")
+ private final AmazonSQSClientMock client = new AmazonSQSClientMock() {
+ @Override
+ public ChangeMessageVisibilityBatchResponse
changeMessageVisibilityBatch(ChangeMessageVisibilityBatchRequest request) {
+ if (request.entries().stream().anyMatch(e ->
SECOND_HANDLE.equals(e.receiptHandle()))) {
+ secondExtended.countDown();
+ }
+ return super.changeMessageVisibilityBatch(request);
+ }
+ };
+
+ {
+ // both messages are on the queue before the route starts, so that the
first poll receives them together
+
client.addMessage(Message.builder().body("first").messageId("id-1").receiptHandle(FIRST_HANDLE).build());
+
client.addMessage(Message.builder().body("second").messageId("id-2").receiptHandle(SECOND_HANDLE).build());
+ }
+
+ @AfterEach
+ void releaseRoute() {
+ releaseFirst.countDown();
+ suspender.shutdownNow();
+ }
+
+ @Test
+ void waitingMessageOfTheBatchIsExtendedWhileTheFirstIsProcessed() throws
Exception {
+ mock.expectedBodiesReceived("first", "second");
+
+ assertThat(firstStarted.await(10, TimeUnit.SECONDS)).as("the first
message was routed").isTrue();
+ // the second message has been received with the first one; it is not
routed before the first is done, but its
+ // visibility timeout is running, so the extender must cover it now
(the first is held for 5 timeouts at most)
+ boolean extended = secondExtended.await(VISIBILITY_TIMEOUT_SECONDS *
5L, TimeUnit.SECONDS);
+ releaseFirst.countDown();
+
+ assertThat(extended)
+ .as("the visibility of the second message of the batch is
extended while the first message is processed"
+ + " (otherwise SQS makes it visible again and delivers it
a second time)")
+ .isTrue();
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void messageLeftInTheBatchIsNoLongerExtendedAfterSuspend() throws
Exception {
+ mock.expectedBodiesReceived("first");
+
+ assertThat(firstStarted.await(10, TimeUnit.SECONDS)).as("the first
message was routed").isTrue();
+ // suspend the route while the first message is processed: the second
message of the batch is then not routed
+ Future<?> suspend = suspender.submit(() -> {
+ context.getRouteController().suspendRoute("sqs", 30,
TimeUnit.SECONDS);
+ return null;
+ });
+ await().atMost(Duration.ofSeconds(10))
+ .until(() -> context.getRoute("sqs").getConsumer() instanceof
Sqs2Consumer c && c.isSuspended());
+ releaseFirst.countDown();
+ suspend.get(30, TimeUnit.SECONDS);
+
assertThat(context.getRouteController().getRouteStatus("sqs")).isEqualTo(ServiceStatus.Suspended);
+ mock.assertIsSatisfied();
+
+ int requestsSoFar =
client.getChangeMessageVisibilityBatchRequests().size();
+ // the message that was not routed must not be kept invisible while
the route is suspended: no extension for
+ // it during several runs of the extender (it runs every
VISIBILITY_TIMEOUT_SECONDS / 2 seconds)
+ await().during(Duration.ofSeconds(VISIBILITY_TIMEOUT_SECONDS *
2L)).atMost(Duration.ofSeconds(10))
+ .until(() ->
client.getChangeMessageVisibilityBatchRequests().stream()
+ .skip(requestsSoFar)
+ .flatMap(r -> r.entries().stream())
+ .noneMatch(e ->
SECOND_HANDLE.equals(e.receiptHandle())));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("aws2-sqs://MyQueue?amazonSQSClient=#amazonSQSClient&maxMessagesPerPoll=2&visibilityTimeout="
+ + VISIBILITY_TIMEOUT_SECONDS +
"&extendMessageVisibility=true")
+ .routeId("sqs")
+ .process(exchange -> {
+ if
("first".equals(exchange.getIn().getBody(String.class))) {
+ firstStarted.countDown();
+ if (!releaseFirst.await(30, TimeUnit.SECONDS))
{
+ throw new IllegalStateException("the test
did not release the first message");
+ }
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+}
