This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/27697-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit e47c8c99d5c5e380aac0fc454ec8fa1c2256c3f4 Author: Guillaume Nodet <[email protected]> AuthorDate: Mon Oct 12 20:47:50 2026 +0000 [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 --- .../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
