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 4452cc0dec4d CAMEL-25809: camel-aws2-sqs - extend the visibility from
the receive, not from the start of the batch (#28201)
4452cc0dec4d is described below
commit 4452cc0dec4d165ba3b2d8ab9eb3b7d0f10b03cb
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 21 13:58:06 2026 +0530
CAMEL-25809: camel-aws2-sqs - extend the visibility from the receive, not
from the start of the batch (#28201)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/aws2/sqs/Sqs2Consumer.java | 56 +++++++--
...onsumerExtendMessageVisibilityLongPollTest.java | 137 +++++++++++++++++++++
2 files changed, 184 insertions(+), 9 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 99cb666cd3ae..2080f3b55ec7 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
@@ -35,6 +35,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@@ -107,13 +108,16 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
shutdownRunningTask = null;
pendingExchanges = 0;
+ // the visibility timeout of a message starts at its receive, and a
poll with several (long polling) receive
+ // requests returns only when the last of them is done, so the
messages are dated from the start of the poll
+ final Instant pollStarted = Instant.now();
List<software.amazon.awssdk.services.sqs.model.Message> messages =
pollingTask.call();
// okay we have some response from aws so lets mark the consumer as
// ready
forceConsumerAsReady();
Queue<Exchange> exchanges = createExchanges(messages);
- return processBatch(CastUtils.cast(exchanges));
+ return processBatch(CastUtils.cast(exchanges), pollStarted);
}
protected Queue<Exchange>
createExchanges(List<software.amazon.awssdk.services.sqs.model.Message>
messages) {
@@ -134,6 +138,10 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
@Override
public int processBatch(Queue<Object> exchanges) throws Exception {
+ return processBatch(exchanges, Instant.now());
+ }
+
+ private int processBatch(Queue<Object> exchanges, Instant received) 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
@@ -141,8 +149,13 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
// again and is delivered a second time
final TimeoutExtender extender = this.timeoutExtender;
if (extender != null) {
+ boolean due = false;
for (Object exchange : exchanges) {
- extender.add((Exchange) exchange);
+ due |= extender.add((Exchange) exchange, received);
+ }
+ if (due) {
+ // a message received long before this batch started may
expire before the next scheduled run
+ extendNow(extender);
}
}
@@ -199,6 +212,18 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
return total;
}
+ private void extendNow(TimeoutExtender extender) {
+ final ScheduledExecutorService executor = this.scheduledExecutor;
+ if (executor != null) {
+ try {
+ // the executor has a single thread, so this run does not
overlap with a scheduled one
+ executor.execute(extender);
+ } catch (RejectedExecutionException e) {
+ LOG.debug("Cannot extend the visibility of the received
messages now, the consumer is stopping", e);
+ }
+ }
+ }
+
private static void stopExtending(TimeoutExtender extender, Queue<Object>
notRouted) {
if (extender != null) {
for (Object exchange : notRouted) {
@@ -419,7 +444,14 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
this.delayBetweenExecutions = delayBetweenExecutions;
}
- public void add(Exchange exchange) {
+ /**
+ * Starts extending the visibility of the message of the given
exchange.
+ *
+ * @param received the time the message was received (or an earlier
time), its visibility timeout runs from
+ * then
+ * @return true if the message must be extended before the
next scheduled run of the extender
+ */
+ public boolean add(Exchange exchange, Instant received) {
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
@Override
public void onComplete(Exchange exchange) {
@@ -441,7 +473,9 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
.id(exchange.getExchangeId()).visibilityTimeout(visibilityTimeout)
.receiptHandle(exchange.getIn().getHeader(Sqs2Constants.RECEIPT_HANDLE,
String.class)).build();
- entries.put(exchange.getExchangeId(), new
TimeoutExtenderEntry(entry));
+ TimeoutExtenderEntry extenderEntry = new
TimeoutExtenderEntry(entry, received);
+ entries.put(exchange.getExchangeId(), extenderEntry);
+ return
extenderEntry.isDeadlineReachedAt(Instant.now().plusSeconds(Math.max(1,
delayBetweenExecutions)));
}
/**
@@ -525,16 +559,20 @@ public class Sqs2Consumer extends
ScheduledBatchPollingConsumer {
*/
private final ChangeMessageVisibilityBatchRequestEntry
extendRequest;
- TimeoutExtenderEntry(ChangeMessageVisibilityBatchRequestEntry
extendRequest) {
+ /**
+ * @param start the time the current visibility timeout of the
message started (its receive or its last
+ * extension)
+ */
+ TimeoutExtenderEntry(ChangeMessageVisibilityBatchRequestEntry
extendRequest, Instant start) {
- // setting deadline to 80% of now until expected visibility
- // timeout, this is for taking into account processing time
- this.deadline =
Instant.now().plusMillis(extendRequest.visibilityTimeout() * 800);
+ // setting deadline to 80% of the visibility timeout after its
+ // start, this is for taking into account processing time
+ this.deadline =
start.plusMillis(extendRequest.visibilityTimeout() * 800L);
this.extendRequest = extendRequest;
}
TimeoutExtenderEntry extendDeadline() {
- return new TimeoutExtenderEntry(extendRequest);
+ return new TimeoutExtenderEntry(extendRequest, Instant.now());
}
boolean isDeadlineReachedAt(Instant time) {
diff --git
a/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsConsumerExtendMessageVisibilityLongPollTest.java
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsConsumerExtendMessageVisibilityLongPollTest.java
new file mode 100644
index 000000000000..eac74530eb9f
--- /dev/null
+++
b/components/camel-aws/camel-aws2-sqs/src/test/java/org/apache/camel/component/aws2/sqs/SqsConsumerExtendMessageVisibilityLongPollTest.java
@@ -0,0 +1,137 @@
+/*
+ * 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.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.EndpointInject;
+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 software.amazon.awssdk.services.sqs.model.ReceiveMessageRequest;
+import software.amazon.awssdk.services.sqs.model.ReceiveMessageResponse;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * With {@code maxMessagesPerPoll} greater than 10 a poll sends several
receive requests and routes the messages only
+ * when the last of them returned, which with long polling can be up to {@code
waitTimeSeconds} after the first one. The
+ * visibility timeout of a message runs from its own receive, so with {@code
extendMessageVisibility=true} its first
+ * extension must come before that receive plus {@code visibilityTimeout}.
+ */
+class SqsConsumerExtendMessageVisibilityLongPollTest extends CamelTestSupport {
+
+ private static final int VISIBILITY_TIMEOUT_SECONDS = 4;
+ // how long the second receive request of the poll waits for messages
(long polling on a queue with no more messages)
+ private static final long LONG_POLL_MILLIS = 3000;
+ private static final String HANDLE = "receipt-handle-1";
+
+ private final AtomicInteger receiveCalls = new AtomicInteger();
+ private final AtomicLong receivedAtNanos = new AtomicLong();
+ private final AtomicLong firstExtensionAtNanos = new AtomicLong();
+ private final CountDownLatch extended = new CountDownLatch(1);
+ private final CountDownLatch releaseRoute = new CountDownLatch(1);
+ private final CountDownLatch endOfTest = new CountDownLatch(1);
+
+ @EndpointInject("mock:result")
+ private MockEndpoint mock;
+
+ @BindToRegistry("amazonSQSClient")
+ private final AmazonSQSClientMock client = new AmazonSQSClientMock() {
+ @Override
+ public ReceiveMessageResponse receiveMessage(ReceiveMessageRequest
request) {
+ if (receiveCalls.incrementAndGet() == 2) {
+ // the other request of the first poll: it waits for messages,
like a long poll on a queue where no
+ // more messages arrive (it returns early only when the test
ends)
+ try {
+ endOfTest.await(LONG_POLL_MILLIS, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ ReceiveMessageResponse response = super.receiveMessage(request);
+ if (response.messages().stream().anyMatch(m ->
HANDLE.equals(m.receiptHandle()))) {
+ // the visibility timeout of the message starts now
+ receivedAtNanos.set(System.nanoTime());
+ }
+ return response;
+ }
+
+ @Override
+ public ChangeMessageVisibilityBatchResponse
changeMessageVisibilityBatch(ChangeMessageVisibilityBatchRequest request) {
+ if (request.entries().stream().anyMatch(e ->
HANDLE.equals(e.receiptHandle()))) {
+ firstExtensionAtNanos.compareAndSet(0, System.nanoTime());
+ extended.countDown();
+ }
+ return super.changeMessageVisibilityBatch(request);
+ }
+ };
+
+ {
+
client.addMessage(Message.builder().body("slow").messageId("id-1").receiptHandle(HANDLE).build());
+ }
+
+ @AfterEach
+ void release() {
+ releaseRoute.countDown();
+ endOfTest.countDown();
+ }
+
+ @Test
+ void firstExtensionComesBeforeTheVisibilityTimeoutOfTheReceiveRunsOut()
throws Exception {
+ mock.expectedBodiesReceived("slow");
+
+ // the route holds the message (a slow processing) until it has been
extended once, at most 3 timeouts
+ boolean wasExtended = extended.await(VISIBILITY_TIMEOUT_SECONDS * 3L,
TimeUnit.SECONDS);
+ releaseRoute.countDown();
+ assertThat(wasExtended).as("the visibility of the message is extended
while it is processed").isTrue();
+
+ long millisFromReceive =
TimeUnit.NANOSECONDS.toMillis(firstExtensionAtNanos.get() -
receivedAtNanos.get());
+ assertThat(millisFromReceive)
+ .as("first ChangeMessageVisibility for the message, in ms
after its receive (it becomes visible again,"
+ + " and is delivered to another consumer, " +
VISIBILITY_TIMEOUT_SECONDS + " s after the receive)")
+ .isLessThan(VISIBILITY_TIMEOUT_SECONDS * 1000L);
+ mock.assertIsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // maxMessagesPerPoll=20: two receive requests of 10 messages
each per poll
+
from("aws2-sqs://MyQueue?amazonSQSClient=#amazonSQSClient&maxMessagesPerPoll=20&visibilityTimeout="
+ + VISIBILITY_TIMEOUT_SECONDS +
"&extendMessageVisibility=true")
+ .process(exchange -> {
+ if (!releaseRoute.await(30, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("the test did
not release the message");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+}
