This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new c6be97386aba [backport camel-4.22.x] CAMEL-25558: camel-aws2-s3 - A
listed object that cannot be fetched no longer blocks the consumer or stays in
progress (#27749)
c6be97386aba is described below
commit c6be97386abaee2b434a69debdecf2c7f71f9aaf
Author: Guillaume Nodet <[email protected]>
AuthorDate: Tue Oct 13 09:40:01 2026 +0200
[backport camel-4.22.x] CAMEL-25558: camel-aws2-s3 - A listed object that
cannot be fetched no longer blocks the consumer or stays in progress (#27749)
Backport of #27697. A listed object that could not be got no longer failed
the
whole poll or stayed in the inProgressRepository: it is skipped and tried
again
by a later poll, its failure goes to the consumer's exception handler, and a
poll in which no listed object could be got still fails. FakeS3Server, the
test
helper the new tests use, is added as on main.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/aws2/s3/AWS2S3Consumer.java | 99 ++++-
...2S3ConsumerDeletedObjectPooledExchangeTest.java | 74 ++++
.../aws2/s3/AWS2S3ConsumerDeletedObjectTest.java | 413 +++++++++++++++++++++
.../camel/component/aws2/s3/FakeS3Server.java | 259 +++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 18 +
5 files changed, 847 insertions(+), 16 deletions(-)
diff --git
a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
index a0e5fce60195..ae01bbe4aa45 100644
---
a/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
+++
b/components/camel-aws/camel-aws2-s3/src/main/java/org/apache/camel/component/aws2/s3/AWS2S3Consumer.java
@@ -19,8 +19,10 @@ package org.apache.camel.component.aws2.s3;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.Queue;
@@ -29,6 +31,7 @@ import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Message;
+import org.apache.camel.PooledExchange;
import org.apache.camel.Processor;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.component.aws2.s3.utils.AWS2S3Utils;
@@ -237,6 +240,10 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
Collection<ResponseInputStream<GetObjectResponse>> s3Objects = new
ArrayList<>();
Queue<Exchange> answer = new LinkedList<>();
+ // the listed objects that could not be got, other than those that no
longer exist, by key
+ Map<String, RuntimeException> failures = new LinkedHashMap<>();
+ // whether S3 answered the get of at least one listed object
+ boolean anyGot = false;
try {
for (S3Object s3ObjectSummary : s3ObjectSummaries) {
String key = s3ObjectSummary.key();
@@ -264,23 +271,37 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
getRequest.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm());
}
}
- ResponseInputStream<GetObjectResponse> s3Object
- = getAmazonS3Client().getObject(getRequest.build(),
ResponseTransformer.toInputStream());
-
- if (includeS3Object(s3Object)) {
- s3Objects.add(s3Object);
- Exchange exchange = createExchange(s3Object, key);
- answer.add(exchange);
- } else {
- // If includeFolders != true and the object is not
included, it is safe to close the object here.
- // If includeFolders == true, the exchange will close the
object.
- IOHelper.close(s3Object);
- // remove in progress
- getEndpoint().getInProgressRepository().remove(key);
+ // a failure to get one listed object, or to create its
exchange, does not fail the poll as long as
+ // another listed object could be got: the object is skipped,
the other listed objects are consumed,
+ // and a later poll tries it again
+ ResponseInputStream<GetObjectResponse> s3Object = null;
+ try {
+ s3Object =
getAmazonS3Client().getObject(getRequest.build(),
ResponseTransformer.toInputStream());
+ if (includeS3Object(s3Object)) {
+ Exchange exchange = createExchange(s3Object, key);
+ s3Objects.add(s3Object);
+ answer.add(exchange);
+ anyGot = true;
+ } else {
+ // If includeFolders != true and the object is not
included, it is safe to close the object
+ // here. If includeFolders == true, the exchange will
close the object.
+ IOHelper.close(s3Object);
+ // remove in progress
+ getEndpoint().getInProgressRepository().remove(key);
+ anyGot = true;
+ }
+ } catch (NoSuchKeyException e) {
+ // deleted after it was listed (by another consumer, a
lifecycle rule, or the exchange of an
+ // earlier poll that completed meanwhile)
+ LOG.debug("Skipping as s3 object no longer exists: {}",
key);
+ skipS3Object(s3Object, key);
+ } catch (RuntimeException e) {
+ skipS3Object(s3Object, key);
+ failures.put(key, e);
}
}
} catch (Exception e) {
- LOG.warn("Error getting S3Object due: {}. This exception is
ignored.", e.getMessage(), e);
+ LOG.warn("Error getting S3Object due: {}", e.getMessage(), e);
// ensure all previous gathered s3 objects are closed
// if there was an exception creating the exchanges in this batch
s3Objects.forEach(IOHelper::close);
@@ -294,9 +315,38 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
throw e;
}
+ if (!failures.isEmpty()) {
+ if (!anyGot) {
+ // none of the listed objects could be got, so this is not
about one object (expired credentials, no
+ // permission to get, throttling, no connection): the poll
fails as it does when the listing fails, and
+ // the error handling and the backoff of the scheduled
consumer apply
+ Map.Entry<String, RuntimeException> first =
failures.entrySet().iterator().next();
+ LOG.warn("None of the listed objects of bucket {} could be got
({} failed, the first is {}): the poll fails",
+ getConfiguration().getBucketName(), failures.size(),
first.getKey());
+ throw first.getValue();
+ }
+ // as processCommit does: the consumer goes on, and a custom
exception handler or bridgeErrorHandler sees
+ // the failure
+ for (Map.Entry<String, RuntimeException> failure :
failures.entrySet()) {
+ String message = "Error getting S3Object " + failure.getKey()
+ " from bucket "
+ + getConfiguration().getBucketName()
+ + ". The object is skipped for this poll.";
+ getExceptionHandler().handleException(message,
failure.getValue());
+ }
+ }
+
return answer;
}
+ /**
+ * The listed object is not consumed by this poll: closes its stream (if
it was got) and removes its key from the
+ * in-progress repository, so that a later poll can consume it.
+ */
+ private void skipS3Object(ResponseInputStream<GetObjectResponse> s3Object,
String key) {
+ IOHelper.close(s3Object);
+ getEndpoint().getInProgressRepository().remove(key);
+ }
+
/**
* Decide whether to include the S3Objects in the results
*
@@ -455,6 +505,25 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
LOG.trace("Got object [{}]", s3Object);
Exchange exchange = createExchange(true);
+ try {
+ populateExchange(exchange, pattern, s3Object, key);
+ } catch (RuntimeException e) {
+ // the object is not consumed (for example its download failed),
so the exchange created for it is not
+ // routed, and nothing else gives it back to the exchange factory
+ if (exchange instanceof PooledExchange pooled) {
+ // created with autoRelease, so done() also gives it back to
the pool, once: releaseExchange would
+ // then add it to the pool a second time, and two later
objects would share the exchange
+ pooled.done();
+ } else {
+ releaseExchange(exchange, true);
+ }
+ throw e;
+ }
+ return exchange;
+ }
+
+ private void populateExchange(
+ Exchange exchange, ExchangePattern pattern,
ResponseInputStream<GetObjectResponse> s3Object, String key) {
exchange.setPattern(pattern);
Message message = exchange.getIn();
@@ -509,8 +578,6 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
});
}
}
-
- return exchange;
}
@Override
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerDeletedObjectPooledExchangeTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerDeletedObjectPooledExchangeTest.java
new file mode 100644
index 000000000000..545457b7555a
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerDeletedObjectPooledExchangeTest.java
@@ -0,0 +1,74 @@
+/*
+ * 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.s3;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.spi.ExchangeFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+/**
+ * The same with the pooled exchange factory: the exchange of an object that
is skipped after it was created (its
+ * download failed) is never routed, and is given back to the pool, once, so
the objects after it do not share it.
+ */
+class AWS2S3ConsumerDeletedObjectPooledExchangeTest extends
AWS2S3ConsumerDeletedObjectTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ PooledExchangeFactory factory = new PooledExchangeFactory();
+ factory.setStatisticsEnabled(true);
+ context.getCamelContextExtension().setExchangeFactory(factory);
+ return context;
+ }
+
+ @Test
+ void exchangeOfAnObjectWhoseDownloadFailsIsGivenBackToThePool() throws
Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.put("c.txt", "c");
+ s3.put("d.txt", "d");
+ // the connection is closed while the content of b.txt is read into
the body
+ s3.cutBodyOfNextGet("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a", "c", "d");
+ AWS2S3Consumer consumer = startRoute("");
+
context.getCamelContextExtension().getExchangeFactoryManager().setStatisticsEnabled(true);
+
+ assertEquals(3, consumer.poll());
+
+ mock.assertIsSatisfied();
+
assertFalse(consumer.getEndpoint().getInProgressRepository().contains("b.txt"),
"b.txt is still in progress");
+ ExchangeFactory.Statistics stats
+ =
context.getCamelContextExtension().getExchangeFactoryManager().getStatistics();
+ // every exchange that was taken (those of a, c and d that were
routed, and the one of b.txt that was not) was
+ // given back, and only once: the pool does not hold an exchange twice
+ assertEquals(4, stats.getCreatedCounter() +
stats.getAcquiredCounter());
+ assertEquals(4, stats.getReleasedCounter());
+ assertEquals(0, stats.getDiscardedCounter());
+
+ // and b.txt is consumed by the next poll
+ mock.reset();
+ mock.expectedBodiesReceived("b");
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerDeletedObjectTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerDeletedObjectTest.java
new file mode 100644
index 000000000000..482f9188834c
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerDeletedObjectTest.java
@@ -0,0 +1,413 @@
+/*
+ * 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.s3;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.core.ResponseInputStream;
+import software.amazon.awssdk.core.sync.ResponseTransformer;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
+import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.model.S3Object;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A listed object that cannot be got (it no longer exists, its get is
refused, its download fails) does not fail the
+ * poll when another listed object could be got: the consumer skips it, gives
the failure to its exception handler
+ * (unless the object no longer exists), consumes the other objects of the
poll, and does not leave it in the
+ * in-progress repository, so a later poll consumes it (or a new object with
the same key). When none of the listed
+ * objects could be got, the poll fails.
+ * <p>
+ * The scheduler of the consumer is not started: the tests call {@code poll()}
themselves.
+ */
+class AWS2S3ConsumerDeletedObjectTest extends CamelTestSupport {
+
+ protected FakeS3Server s3;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ s3 = new FakeS3Server("bkt");
+ s3.start();
+ return super.createCamelContext();
+ }
+
+ @AfterEach
+ void stopServer() {
+ if (s3 != null) {
+ s3.stop();
+ }
+ }
+
+ private String s3Uri(String options) {
+ return
"aws2-s3://bkt?accessKey=xxx&secretKey=yyy®ion=us-east-1&overrideEndpoint=true&forcePathStyle=true"
+ + "&uriEndpointOverride=" + s3.endpoint() +
"&startScheduler=false" + options;
+ }
+
+ protected AWS2S3Consumer startRoute(String options) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(s3Uri(options)).routeId("s3")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+ return (AWS2S3Consumer) context.getRoute("s3").getConsumer();
+ }
+
+ @Test
+ void listedObjectThatNoLongerExistsIsSkipped() throws Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.put("c.txt", "c");
+ // deleted after the listing has returned it and before the consumer
gets it
+ s3.removeWhenListed("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a", "c");
+ mock.expectedPropertyReceived(Exchange.BATCH_SIZE, 2);
+ AWS2S3Consumer consumer = startRoute("");
+ RecordingExceptionHandler handler = new RecordingExceptionHandler();
+ consumer.setExceptionHandler(handler);
+
+ int polled = consumer.poll();
+
+ // the object that is gone is not counted either
+ assertEquals(2, polled);
+ mock.assertIsSatisfied();
+ // and it is not an error
+ assertTrue(handler.messages.isEmpty(), "exception handler was called:
" + handler.messages);
+
assertFalse(consumer.getEndpoint().getInProgressRepository().contains("b.txt"),
"b.txt is still in progress");
+ }
+
+ @Test
+ void objectCreatedAgainAfterItWasSkippedIsConsumed() throws Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.removeWhenListed("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a");
+ AWS2S3Consumer consumer = startRoute("&includeBody=false");
+
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+
+ // a new object with the key of the one that was skipped
+ mock.reset();
+ mock.expectedBodiesReceived("b2");
+ s3.put("b.txt", "b2");
+
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void objectWhoseGetIsRefusedDoesNotStopTheOtherObjects() throws Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.put("c.txt", "c");
+ // as an object that cannot be read at all (archived, access denied):
every get fails
+ s3.refuseGets("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a", "c");
+ mock.expectedPropertyReceived(Exchange.BATCH_SIZE, 2);
+ AWS2S3Consumer consumer = startRoute("");
+ RecordingExceptionHandler handler = new RecordingExceptionHandler();
+ consumer.setExceptionHandler(handler);
+
+ // the poll does not fail, and the objects listed before and after the
refused one are consumed
+ assertEquals(2, consumer.poll());
+ mock.assertIsSatisfied();
+ assertEquals(1, s3.refusedGets("b.txt"));
+ // the exception handler of the consumer got the failure of the
skipped object
+ assertEquals(1, handler.exceptions.size());
+ S3Exception refused = assertInstanceOf(S3Exception.class,
handler.exceptions.get(0));
+ assertEquals(403, refused.statusCode());
+ assertTrue(handler.messages.get(0).contains("b.txt"),
handler.messages.get(0));
+
assertFalse(consumer.getEndpoint().getInProgressRepository().contains("b.txt"),
"b.txt is still in progress");
+
+ // the following poll consumes a new object, and tries the refused one
again
+ mock.reset();
+ mock.expectedBodiesReceived("d");
+ s3.put("d.txt", "d");
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+ assertEquals(2, s3.refusedGets("b.txt"));
+
+ // and the object is consumed once it can be got
+ mock.reset();
+ mock.expectedBodiesReceived("b");
+ s3.allowGets("b.txt");
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void
skippedObjectOfATruncatedListingIsTriedAgainWhenTheListingStartsOver() throws
Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.put("c.txt", "c");
+ s3.put("d.txt", "d");
+ s3.refuseGets("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a");
+ AWS2S3Consumer consumer = startRoute("&maxMessagesPerPoll=2");
+
+ // first page (a, b): the skipped object is not counted
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+
+ // the marker has moved on as for any page: the second page (c, d) is
consumed, and b is not tried
+ mock.reset();
+ mock.expectedBodiesReceived("c", "d");
+ assertEquals(2, consumer.poll());
+ mock.assertIsSatisfied();
+ assertEquals(1, s3.refusedGets("b.txt"));
+
+ // the listing starts over: b is tried again (with e.txt, as a poll in
which no object could be got fails)
+ mock.reset();
+ mock.expectedBodiesReceived("e");
+ s3.put("e.txt", "e");
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+ assertEquals(2, s3.refusedGets("b.txt"));
+
+ mock.reset();
+ mock.expectedBodiesReceived("b");
+ s3.allowGets("b.txt");
+ assertEquals(1, consumer.poll());
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void pollFailsWhenNoneOfTheListedObjectsCouldBeGot() throws Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ // as with expired credentials or no permission to get: the listing
works and every get fails
+ s3.refuseGets("a.txt");
+ s3.refuseGets("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(0);
+ AWS2S3Consumer consumer = startRoute("");
+ RecordingExceptionHandler handler = new RecordingExceptionHandler();
+ consumer.setExceptionHandler(handler);
+
+ // the poll fails with the failure of the first object, so the
scheduled consumer counts an error (backoff)
+ S3Exception e = assertThrows(S3Exception.class, consumer::poll);
+ assertEquals(403, e.statusCode());
+ mock.assertIsSatisfied();
+ // every listed object was tried, and none is left in progress
+ assertEquals(1, s3.refusedGets("a.txt"));
+ assertEquals(1, s3.refusedGets("b.txt"));
+
assertFalse(consumer.getEndpoint().getInProgressRepository().contains("a.txt"),
"a.txt is still in progress");
+
assertFalse(consumer.getEndpoint().getInProgressRepository().contains("b.txt"),
"b.txt is still in progress");
+ // the failure is the failure of the poll, and is not given to the
exception handler per object as well
+ assertTrue(handler.messages.isEmpty(), "exception handler was called:
" + handler.messages);
+
+ // the next poll tries the objects again
+ mock.reset();
+ mock.expectedBodiesReceived("a", "b");
+ s3.allowGets("a.txt");
+ s3.allowGets("b.txt");
+ assertEquals(2, consumer.poll());
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void pollDoesNotFailWhenTheOnlyListedObjectNoLongerExists() throws
Exception {
+ s3.put("a.txt", "a");
+ s3.removeWhenListed("a.txt");
+ AWS2S3Consumer consumer = startRoute("");
+
+ assertEquals(0, consumer.poll());
+ }
+
+ @Test
+ void failureOfASkippedObjectIsRoutedWithBridgeErrorHandler() throws
Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.refuseGets("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a");
+ MockEndpoint error = getMockEndpoint("mock:error");
+ error.expectedMessageCount(1);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ onException(S3Exception.class).handled(true).to("mock:error");
+
+ from(s3Uri("&bridgeErrorHandler=true")).routeId("s3")
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+ AWS2S3Consumer consumer = (AWS2S3Consumer)
context.getRoute("s3").getConsumer();
+
+ assertEquals(1, consumer.poll());
+
+ mock.assertIsSatisfied();
+ error.assertIsSatisfied();
+ String message =
error.getReceivedExchanges().get(0).getMessage().getBody(String.class);
+ assertTrue(message.contains("b.txt"), message);
+ }
+
+ @Test
+ void objectWhoseDownloadFailsIsSkippedAndNotLeftInProgress() throws
Exception {
+ AWS2S3Endpoint endpoint = context.getEndpoint(s3Uri(""),
AWS2S3Endpoint.class);
+ AtomicBoolean closed = new AtomicBoolean();
+ // the download of the object fails while its content is read into the
body (includeBody=true)
+ InputStream content = new InputStream() {
+ @Override
+ public int read() throws IOException {
+ throw new IOException("connection reset");
+ }
+
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ };
+ AWS2S3Consumer consumer = new AWS2S3Consumer(endpoint, exchange -> {
+ }) {
+ @Override
+ protected S3Client getAmazonS3Client() {
+ return oneObject("a.txt", content);
+ }
+ };
+ // not started, but the exchange factory of the consumer must be ready
+ consumer.build();
+
+ // it is the only listed object, so the poll fails
+ RuntimeCamelException e = assertThrows(RuntimeCamelException.class,
consumer::poll);
+ assertInstanceOf(IOException.class, e.getCause());
+
+ assertTrue(closed.get(), "the object stream of a.txt was not closed");
+ assertFalse(endpoint.getInProgressRepository().contains("a.txt"),
"a.txt is still in progress");
+ }
+
+ @Test
+ void objectStreamIsClosedWhenIncludeS3ObjectFails() throws Exception {
+ AWS2S3Endpoint endpoint = context.getEndpoint(s3Uri(""),
AWS2S3Endpoint.class);
+ AtomicBoolean closed = new AtomicBoolean();
+ InputStream content = new InputStream() {
+ @Override
+ public int read() {
+ return -1;
+ }
+
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ };
+ AWS2S3Consumer consumer = new AWS2S3Consumer(endpoint, exchange -> {
+ }) {
+ @Override
+ protected S3Client getAmazonS3Client() {
+ return oneObject("a.txt", content);
+ }
+
+ @Override
+ protected boolean
includeS3Object(ResponseInputStream<GetObjectResponse> s3Object) {
+ throw new IllegalStateException("cannot decide");
+ }
+ };
+ // not started, but the exchange factory of the consumer must be ready
+ consumer.build();
+
+ // it is the only listed object, so the poll fails
+ assertThrows(IllegalStateException.class, consumer::poll);
+
+ assertTrue(closed.get(), "the object stream of a.txt was not closed");
+ assertFalse(endpoint.getInProgressRepository().contains("a.txt"),
"a.txt is still in progress");
+ }
+
+ /**
+ * Records what the consumer gives to its exception handler.
+ */
+ private static final class RecordingExceptionHandler implements
ExceptionHandler {
+ final List<String> messages = new CopyOnWriteArrayList<>();
+ final List<Throwable> exceptions = new CopyOnWriteArrayList<>();
+
+ @Override
+ public void handleException(Throwable exception) {
+ handleException(null, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ handleException(message, null, exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ messages.add(String.valueOf(message));
+ exceptions.add(exception);
+ }
+ }
+
+ /**
+ * A client whose listing has the one object, and whose get returns the
content.
+ */
+ private static S3Client oneObject(String key, InputStream content) {
+ return new S3Client() {
+ @Override
+ public ListObjectsV2Response listObjectsV2(ListObjectsV2Request
request) {
+ return
ListObjectsV2Response.builder().contents(S3Object.builder().key(key).build())
+ .isTruncated(false).build();
+ }
+
+ @Override
+ public <T> T getObject(GetObjectRequest request,
ResponseTransformer<GetObjectResponse, T> transformer) {
+ @SuppressWarnings("unchecked")
+ T answer = (T) new
ResponseInputStream<>(GetObjectResponse.builder().eTag("tag").build(), content);
+ return answer;
+ }
+
+ @Override
+ public String serviceName() {
+ return SERVICE_NAME;
+ }
+
+ @Override
+ public void close() {
+ // noop
+ }
+ };
+ }
+}
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/FakeS3Server.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/FakeS3Server.java
new file mode 100644
index 000000000000..c9869dc581b0
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/FakeS3Server.java
@@ -0,0 +1,259 @@
+/*
+ * 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.s3;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * A minimal in-process fake of the S3 API calls the consumer makes for a
single bucket (path-style requests): head
+ * bucket, list objects (v2, paginated with max-keys and
start-after/continuation-token), head, get and delete object.
+ * An object can be removed right after it was listed, the gets of an object
can be refused, and the body of an object
+ * can be cut short.
+ */
+final class FakeS3Server {
+
+ private final String bucket;
+ private final HttpServer server;
+ private final NavigableMap<String, byte[]> objects = new
ConcurrentSkipListMap<>();
+ private final Map<String, AtomicInteger> gets = new ConcurrentHashMap<>();
+ private final Set<String> removedWhenListed =
ConcurrentHashMap.newKeySet();
+ private final Set<String> refused = ConcurrentHashMap.newKeySet();
+ private final Map<String, AtomicInteger> refusedGets = new
ConcurrentHashMap<>();
+ private final Set<String> cutBodies = ConcurrentHashMap.newKeySet();
+
+ FakeS3Server(String bucket) throws IOException {
+ this.bucket = bucket;
+ this.server = HttpServer.create(new InetSocketAddress("localhost", 0),
0);
+ this.server.createContext("/", this::handle);
+ }
+
+ void start() {
+ server.start();
+ }
+
+ void stop() {
+ server.stop(0);
+ }
+
+ String endpoint() {
+ return "http://localhost:"; + server.getAddress().getPort();
+ }
+
+ void put(String key, String content) {
+ objects.put(key, content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ /**
+ * The object is removed as soon as a listing has returned it (once), as
another consumer or a lifecycle rule does.
+ */
+ void removeWhenListed(String key) {
+ removedWhenListed.add(key);
+ }
+
+ /**
+ * Every get of the object is refused with 403 (AccessDenied), which the
client does not retry, until
+ * {@link #allowGets(String)}.
+ */
+ void refuseGets(String key) {
+ refused.add(key);
+ }
+
+ void allowGets(String key) {
+ refused.remove(key);
+ }
+
+ /**
+ * The number of gets of the object that were refused.
+ */
+ int refusedGets(String key) {
+ AtomicInteger n = refusedGets.get(key);
+ return n != null ? n.get() : 0;
+ }
+
+ /**
+ * The connection is closed before the whole body of the object was sent
(once), so its download fails.
+ */
+ void cutBodyOfNextGet(String key) {
+ cutBodies.add(key);
+ }
+
+ int gets(String key) {
+ AtomicInteger n = gets.get(key);
+ return n != null ? n.get() : 0;
+ }
+
+ private void handle(HttpExchange http) throws IOException {
+ String path = http.getRequestURI().getRawPath();
+ String key = path.length() > bucket.length() + 2 ?
decode(path.substring(bucket.length() + 2)) : null;
+ String method = http.getRequestMethod();
+ if (key == null) {
+ if ("GET".equals(method) && query(http).containsKey("list-type")) {
+ list(http, query(http));
+ } else {
+ // head bucket, location, ...
+ respond(http, 200, null, null);
+ }
+ return;
+ }
+ byte[] data = objects.get(key);
+ switch (method) {
+ case "HEAD" -> {
+ if (data == null) {
+ respond(http, 404, null, null);
+ } else {
+ objectHeaders(http, data);
+ http.sendResponseHeaders(200, -1);
+ http.close();
+ }
+ }
+ case "GET" -> {
+ if (refused.contains(key)) {
+ refusedGets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
+ error(http, 403, "AccessDenied", "Access Denied", key);
+ } else if (data == null) {
+ noSuchKey(http, key);
+ } else {
+ gets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
+ objectHeaders(http, data);
+ if (cutBodies.remove(key)) {
+ // announces more than is sent, and closes the
connection
+ http.sendResponseHeaders(200, data.length + 100L);
+ http.getResponseBody().write(data);
+ http.getResponseBody().flush();
+ http.close();
+ return;
+ }
+ http.sendResponseHeaders(200, data.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(data);
+ }
+ }
+ }
+ case "DELETE" -> {
+ objects.remove(key);
+ http.sendResponseHeaders(204, -1);
+ http.close();
+ }
+ default -> respond(http, 405, null, null);
+ }
+ }
+
+ private void list(HttpExchange http, Map<String, String> query) throws
IOException {
+ int maxKeys = query.containsKey("max-keys") ?
Integer.parseInt(query.get("max-keys")) : 1000;
+ String after = query.get("continuation-token");
+ if (after == null) {
+ after = query.get("start-after");
+ }
+ NavigableMap<String, byte[]> view = after != null ?
objects.tailMap(after, false) : objects;
+ StringBuilder contents = new StringBuilder();
+ int count = 0;
+ String last = null;
+ List<String> listed = new ArrayList<>();
+ for (Map.Entry<String, byte[]> e : view.entrySet()) {
+ if (count == maxKeys) {
+ break;
+ }
+
contents.append("<Contents><Key>").append(e.getKey()).append("</Key>")
+
.append("<LastModified>2026-10-09T00:00:00.000Z</LastModified>")
+
.append("<ETag>\"0123456789abcdef0123456789abcdef\"</ETag>")
+
.append("<Size>").append(e.getValue().length).append("</Size>")
+
.append("<StorageClass>STANDARD</StorageClass></Contents>");
+ last = e.getKey();
+ listed.add(last);
+ count++;
+ }
+ boolean truncated = last != null && objects.higherKey(last) != null;
+ String xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
+ + "<ListBucketResult
xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\";>"
+ + "<Name>" + bucket + "</Name><Prefix></Prefix>"
+ + "<KeyCount>" + count + "</KeyCount><MaxKeys>" + maxKeys
+ "</MaxKeys>"
+ + "<IsTruncated>" + truncated + "</IsTruncated>"
+ + (truncated ? "<NextContinuationToken>" + last +
"</NextContinuationToken>" : "")
+ + contents + "</ListBucketResult>";
+ respond(http, 200, "application/xml", xml);
+ for (String key : listed) {
+ if (removedWhenListed.remove(key)) {
+ objects.remove(key);
+ }
+ }
+ }
+
+ private static void objectHeaders(HttpExchange http, byte[] data) {
+ http.getResponseHeaders().add("ETag",
"\"0123456789abcdef0123456789abcdef\"");
+ http.getResponseHeaders().add("Last-Modified", "Fri, 09 Oct 2026
00:00:00 GMT");
+ http.getResponseHeaders().add("Content-Type", "text/plain");
+ http.getResponseHeaders().add("Content-Length",
String.valueOf(data.length));
+ }
+
+ private void noSuchKey(HttpExchange http, String key) throws IOException {
+ error(http, 404, "NoSuchKey", "The specified key does not exist.",
key);
+ }
+
+ private void error(HttpExchange http, int status, String code, String
message, String key) throws IOException {
+ String xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><Error><Code>"
+ code + "</Code>"
+ + "<Message>" + message + "</Message><Key>" + key +
"</Key>"
+ + "<BucketName>" + bucket + "</BucketName><Resource>/" +
bucket + "/" + key + "</Resource>"
+ + "<RequestId>1</RequestId><HostId>1</HostId></Error>";
+ respond(http, status, "application/xml", xml);
+ }
+
+ private static void respond(HttpExchange http, int status, String
contentType, String body) throws IOException {
+ if (body == null) {
+ http.sendResponseHeaders(status, -1);
+ http.close();
+ return;
+ }
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ http.getResponseHeaders().add("Content-Type", contentType);
+ http.sendResponseHeaders(status, bytes.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(bytes);
+ }
+ }
+
+ private static Map<String, String> query(HttpExchange http) {
+ Map<String, String> answer = new HashMap<>();
+ String q = http.getRequestURI().getRawQuery();
+ if (q != null) {
+ for (String part : q.split("&")) {
+ int eq = part.indexOf('=');
+ answer.put(decode(eq >= 0 ? part.substring(0, eq) : part), eq
>= 0 ? decode(part.substring(eq + 1)) : "");
+ }
+ }
+ return answer;
+ }
+
+ private static String decode(String s) {
+ return URLDecoder.decode(s, StandardCharsets.UTF_8);
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
index 096158878345..c43d1d8e7c24 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc
@@ -80,6 +80,24 @@ Not changed: the HMAC key is still derived from the same key
material as the cip
would change the MAC written into the message and so could not be read by
earlier versions; that is
tracked separately.
+=== camel-aws2-s3 - a listed object that cannot be fetched is skipped
+
+When the consumer lists the bucket, a failure to get one of the listed objects
(or to download its content) no longer
+fails the poll, as long as another object of that listing could be got: the
object is skipped, the other listed
+objects are consumed in that poll, and the skipped object is tried again by a
later poll. An object that no longer
+exists (for example deleted by another consumer after the listing) is skipped
with a log line at `DEBUG`. Any other
+failure is given to the exception handler of the consumer, with the key in the
message: by default that is a `WARN`
+with the stack trace, and with `bridgeErrorHandler=true` the route's error
handler gets it. This happens at every poll
+that lists an object that can never be read (for example archived, or access
denied), until it can be read or is
+removed from the bucket. Such a poll is a successful poll, so it does not
count for `backoffErrorThreshold`.
+
+The poll still fails, as before, when none of the listed objects could be got
and at least one get failed with an
+error other than that the object no longer exists (for example expired
credentials, no permission to get objects,
+throttling): the failure of the first object is then the failure of the poll,
and `backoffErrorThreshold` applies.
+In both cases the objects are no longer kept in the `inProgressRepository`;
before, the object (or a new object with
+the same key) was skipped by all later polls until the endpoint was restarted.
With the `fileName` option, a failed
+get still fails the poll.
+
== Upgrading from 4.22.0 to 4.22.1
=== camel-hazelcast - ReplicatedHazelcastAggregationRepository now applies the
default serialization filter
