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 91e523dafe45 CAMEL-25521: camel-aws2-s3 - Keep the in-progress
repository of the consumer accurate with fileName and on stop (#27672)
91e523dafe45 is described below
commit 91e523dafe457ce6be9f9a1841adf56a7b985eff
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 11 13:25:03 2026 +0530
CAMEL-25521: camel-aws2-s3 - Keep the in-progress repository of the
consumer accurate with fileName and on stop (#27672)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/aws2/s3/AWS2S3Consumer.java | 71 +++--
.../aws2/s3/AWS2S3ConsumerInProgressTest.java | 313 +++++++++++++++++++++
.../camel/component/aws2/s3/FakeS3Server.java | 194 +++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 7 +
4 files changed, 568 insertions(+), 17 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..57c114127251 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
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.aws2.s3;
+import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
@@ -124,23 +125,23 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
if (!doneFileCheckPasses(bucketName, doneFileName)) {
exchanges = new LinkedList<>();
} else if (fileName != null) {
- LOG.trace("Getting object in bucket [{}] with file name [{}]...",
bucketName, fileName);
-
- GetObjectRequest.Builder getRequest =
GetObjectRequest.builder().bucket(bucketName).key(fileName);
- if (getConfiguration().isUseCustomerKey()) {
- if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyId())) {
-
getRequest.sseCustomerKey(getConfiguration().getCustomerKeyId());
- }
- if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyMD5())) {
-
getRequest.sseCustomerKeyMD5(getConfiguration().getCustomerKeyMD5());
- }
- if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerAlgorithm())) {
-
getRequest.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm());
+ // check if file is already in progress (add false = duplicate
file)
+ if (!getEndpoint().getInProgressRepository().add(fileName)) {
+ LOG.trace("Skipping as s3 object is already in progress: {}",
fileName);
+ exchanges = new LinkedList<>();
+ } else {
+ ResponseInputStream<GetObjectResponse> s3Object = null;
+ try {
+ s3Object = getObject(bucketName, fileName);
+ exchanges = createExchanges(s3Object, fileName);
+ } catch (Exception e) {
+ // close the stream to release the underlying HTTP
connection
+ IOHelper.close(s3Object);
+ // remove in progress as we failed
+ getEndpoint().getInProgressRepository().remove(fileName);
+ throw e;
}
}
- ResponseInputStream<GetObjectResponse> s3Object
- = getAmazonS3Client().getObject(getRequest.build());
- exchanges = createExchanges(s3Object, fileName);
} else {
LOG.trace("Queueing objects in bucket [{}]...", bucketName);
@@ -203,6 +204,24 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
return processBatch(CastUtils.cast(exchanges));
}
+ private ResponseInputStream<GetObjectResponse> getObject(String
bucketName, String fileName) {
+ LOG.trace("Getting object in bucket [{}] with file name [{}]...",
bucketName, fileName);
+
+ GetObjectRequest.Builder getRequest =
GetObjectRequest.builder().bucket(bucketName).key(fileName);
+ if (getConfiguration().isUseCustomerKey()) {
+ if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyId())) {
+
getRequest.sseCustomerKey(getConfiguration().getCustomerKeyId());
+ }
+ if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerKeyMD5())) {
+
getRequest.sseCustomerKeyMD5(getConfiguration().getCustomerKeyMD5());
+ }
+ if
(ObjectHelper.isNotEmpty(getConfiguration().getCustomerAlgorithm())) {
+
getRequest.sseCustomerAlgorithm(getConfiguration().getCustomerAlgorithm());
+ }
+ }
+ return getAmazonS3Client().getObject(getRequest.build());
+ }
+
private boolean doneFileCheckPasses(String bucketName, String
doneFileName) {
if (doneFileName == null) {
return true;
@@ -290,6 +309,8 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
if (key != null) {
getEndpoint().getInProgressRepository().remove(key);
}
+ // the exchange is never routed, so nothing else releases it
to the exchange factory
+ releaseExchange(exchange, true);
}
throw e;
}
@@ -361,6 +382,21 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+ // the remaining exchanges are not processed as the consumer is
stopping: a later poll must consume their
+ // objects again, and with includeBody=false their bodies are the
object streams, which nothing else closes
+ for (Object notRouted : exchanges) {
+ Exchange exchange = (Exchange) notRouted;
+ String key = exchange.getProperty(AWS2S3Constants.KEY,
String.class);
+ if (key != null) {
+ getEndpoint().getInProgressRepository().remove(key);
+ }
+ if (exchange.getIn().getBody() instanceof Closeable body) {
+ IOHelper.close(body);
+ }
+ // nothing else releases an exchange that is never routed to the
exchange factory (which resets a pooled
+ // exchange, so this comes last)
+ releaseExchange(exchange, true);
+ }
return total;
}
@@ -493,11 +529,12 @@ public class AWS2S3Consumer extends
ScheduledBatchPollingConsumer {
/*
* If includeBody == true, it is safe to close the object here because
the S3Object
- * was consumed already. If includeBody != true, the caller is
responsible for
+ * was consumed already. The same goes for ignoreBody == true, as the
stream is then not
+ * the body, so nothing else can read or close it. Otherwise the
caller is responsible for
* closing the stream once the body has been fully consumed or use the
autoCloseBody
* configuration to automatically schedule the body closing at the end
of exchange.
*/
- if (getConfiguration().isIncludeBody()) {
+ if (getConfiguration().isIncludeBody() ||
getConfiguration().isIgnoreBody()) {
IOHelper.close(s3Object);
} else {
if (getConfiguration().isAutocloseBody()) {
diff --git
a/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerInProgressTest.java
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerInProgressTest.java
new file mode 100644
index 000000000000..6b7259dec654
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/AWS2S3ConsumerInProgressTest.java
@@ -0,0 +1,313 @@
+/*
+ * 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.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+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.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.GetObjectResponse;
+
+import static org.awaitility.Awaitility.await;
+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;
+
+/**
+ * The consumer keeps the objects whose exchanges are being processed in its
in-progress repository: with fileName, an
+ * object still in flight is not consumed again by the next poll, a failed get
does not leave the object in progress,
+ * and the objects of a batch that a stopping consumer does not process are
consumed when the route is started again,
+ * and their object streams are closed and their exchanges released. With
ignoreBody=true the object stream is closed
+ * when the exchange is created, and with fileName it is closed when the
exchange cannot be created.
+ */
+class AWS2S3ConsumerInProgressTest extends CamelTestSupport {
+
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch failedPoll = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private FakeS3Server s3;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ s3 = new FakeS3Server("bkt");
+ s3.start();
+ CamelContext context = super.createCamelContext();
+ // counts the polls that completed (the exchanges of the poll were
handed to the route)
+ context.getRegistry().bind("countPolls", new
PollingConsumerPollStrategy() {
+ @Override
+ public boolean begin(Consumer consumer, Endpoint endpoint) {
+ return true;
+ }
+
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ twoPolls.countDown();
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ failedPoll.countDown();
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void stopServer() {
+ release.countDown();
+ 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() + "&delay=10" +
options;
+ }
+
+ @Test
+ void objectInFlightIsNotConsumedAgainWithFileName() throws Exception {
+ s3.put("a.txt", "hello");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(s3Uri("&fileName=a.txt&pollStrategy=#countPolls"))
+ .to("seda:hold");
+
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:hold")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ assertTrue(inRoute.await(20, TimeUnit.SECONDS), "the object was not
consumed");
+ // the second poll has completed while the first exchange is still
held in the seda route
+ assertTrue(twoPolls.await(20, TimeUnit.SECONDS), "the consumer did not
poll again");
+ int gets = s3.gets("a.txt");
+ release.countDown();
+
+ assertEquals(1, gets, "a.txt was consumed again by a later poll while
its first exchange was still in flight");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void objectWithFileNameIsConsumedOnceItExists() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(s3Uri("&fileName=a.txt&pollStrategy=#countPolls"))
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ // the get fails (NoSuchKey) while the object does not exist, which
must not leave it in progress
+ assertTrue(failedPoll.await(20, TimeUnit.SECONDS), "the get of the
missing object did not fail");
+ s3.put("a.txt", "hello");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void objectsOfAStoppedBatchAreConsumedAfterRestart() throws Exception {
+ s3.put("a.txt", "a");
+ s3.put("b.txt", "b");
+ s3.put("c.txt", "c");
+ AtomicBoolean first = new AtomicBoolean(true);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(s3Uri("")).routeId("s3")
+ .process(e -> {
+ if (first.compareAndSet(true, false)) {
+ // stop the route while the first exchange of
the batch is processed, and wait until
+ // the consumer does not process the rest of
the batch
+ AWS2S3Consumer consumer = (AWS2S3Consumer)
context.getRoute("s3").getConsumer();
+ CompletableFuture.runAsync(() -> {
+ try {
+
context.getRouteController().stopRoute("s3");
+ } catch (Exception ex) {
+ throw new RuntimeCamelException(ex);
+ }
+ });
+ await().atMost(20, TimeUnit.SECONDS).until(()
-> !consumer.isBatchAllowed());
+ }
+ })
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ mock.expectedBodiesReceived("a");
+ mock.assertIsSatisfied();
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() -> context.getRouteController().getRouteStatus("s3")
== ServiceStatus.Stopped);
+
+ mock.reset();
+ mock.expectedBodiesReceivedInAnyOrder("b", "c");
+ context.getRouteController().startRoute("s3");
+
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void notProcessedExchangesReleaseTheirObjects() throws Exception {
+ AWS2S3Endpoint endpoint =
context.getEndpoint(s3Uri("&includeBody=false"), AWS2S3Endpoint.class);
+ List<Boolean> closedWhenReleased = new ArrayList<>();
+ AtomicBoolean closed = new AtomicBoolean();
+ // never started, so the consumer does not process any exchange of the
batch
+ AWS2S3Consumer consumer = new AWS2S3Consumer(endpoint, exchange -> {
+ }) {
+ @Override
+ public void releaseExchange(Exchange exchange, boolean
autoRelease) {
+ closedWhenReleased.add(closed.get());
+ super.releaseExchange(exchange, autoRelease);
+ }
+ };
+ endpoint.getInProgressRepository().add("b.txt");
+ Exchange exchange = endpoint.createExchange();
+ exchange.setProperty(AWS2S3Constants.KEY, "b.txt");
+ exchange.getIn().setBody(new
ByteArrayInputStream("b".getBytes(StandardCharsets.UTF_8)) {
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ });
+ Queue<Object> batch = new LinkedList<>();
+ batch.add(exchange);
+
+ consumer.processBatch(batch);
+
+ assertFalse(endpoint.getInProgressRepository().contains("b.txt"),
"b.txt is still in progress");
+ assertTrue(closed.get(), "the object stream of b.txt was not closed");
+ // released to the exchange factory, after the stream was closed as a
pooled exchange is reset when released
+ assertEquals(List.of(true), closedWhenReleased, "the exchange of b.txt
was not released after closing its stream");
+ }
+
+ @Test
+ void ignoredBodyStreamIsClosed() throws Exception {
+ AWS2S3Endpoint endpoint =
context.getEndpoint(s3Uri("&includeBody=false&ignoreBody=true"),
AWS2S3Endpoint.class);
+ AWS2S3Consumer consumer = (AWS2S3Consumer)
endpoint.createConsumer(exchange -> {
+ });
+ AtomicBoolean closed = new AtomicBoolean();
+ InputStream content = new
ByteArrayInputStream("b".getBytes(StandardCharsets.UTF_8)) {
+ @Override
+ public void close() {
+ closed.set(true);
+ }
+ };
+
+ Exchange exchange = consumer.createExchange(
+ new
ResponseInputStream<>(GetObjectResponse.builder().eTag("tag").build(),
content), "b.txt");
+
+ // the stream is not the body, so nothing else can close it (an
exchange that is never routed does not even
+ // run the on completion of autocloseBody)
+ assertTrue(closed.get(), "the object stream of b.txt was not closed");
+ assertEquals("tag", exchange.getIn().getHeader(AWS2S3Constants.E_TAG));
+ assertEquals("b.txt", exchange.getIn().getHeader(AWS2S3Constants.KEY));
+ }
+
+ @Test
+ void objectStreamIsClosedWhenTheExchangeOfAFileNamePollCannotBeCreated()
throws Exception {
+ AWS2S3Endpoint endpoint =
context.getEndpoint(s3Uri("&fileName=a.txt"), 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);
+ }
+ };
+ S3Client client = new S3Client() {
+ @Override
+ public ResponseInputStream<GetObjectResponse>
getObject(GetObjectRequest request) {
+ return new
ResponseInputStream<>(GetObjectResponse.builder().eTag("tag").build(), content);
+ }
+
+ @Override
+ public String serviceName() {
+ return SERVICE_NAME;
+ }
+
+ @Override
+ public void close() {
+ // noop
+ }
+ };
+ AWS2S3Consumer consumer = new AWS2S3Consumer(endpoint, exchange -> {
+ }) {
+ @Override
+ protected S3Client getAmazonS3Client() {
+ return client;
+ }
+ };
+
+ RuntimeCamelException e = assertThrows(RuntimeCamelException.class,
consumer::poll);
+
+ assertInstanceOf(IOException.class, e.getCause());
+ assertEquals("connection reset", e.getCause().getMessage());
+ assertFalse(endpoint.getInProgressRepository().contains("a.txt"),
"a.txt is still in progress");
+ // nothing else has a reference to the stream, so its connection would
never be released
+ assertTrue(closed.get(), "the object stream of a.txt was not closed");
+ }
+}
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..1e2b4aa4ea26
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3/src/test/java/org/apache/camel/component/aws2/s3/FakeS3Server.java
@@ -0,0 +1,194 @@
+/*
+ * 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.HashMap;
+import java.util.Map;
+import java.util.NavigableMap;
+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.
+ */
+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<>();
+
+ 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));
+ }
+
+ 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 (data == null) {
+ noSuchKey(http, key);
+ } else {
+ gets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
+ objectHeaders(http, data);
+ 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;
+ 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();
+ 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);
+ }
+
+ 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 {
+ String xml = "<?xml version=\"1.0\"
encoding=\"UTF-8\"?><Error><Code>NoSuchKey</Code>"
+ + "<Message>The specified key does not
exist.</Message><Key>" + key + "</Key>"
+ + "<BucketName>" + bucket + "</BucketName><Resource>/" +
bucket + "/" + key + "</Resource>"
+ + "<RequestId>1</RequestId><HostId>1</HostId></Error>";
+ respond(http, 404, "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_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index ced580f92f30..ae3640b4cef0 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -615,6 +615,13 @@ A header that carries a literal `${...}` string is now
used as the object key ve
A configured `keyName` or `bucketName` whose Simple expression resolves to
`null` now fails with an
`IllegalArgumentException` at the producer, instead of passing a `null`
key/bucket on to the AWS SDK.
+=== camel-aws2-s3 - objects in progress in the consumer
+
+With `fileName` set, the consumer now skips the object while an exchange of an
earlier poll is still processing it, as
+it already did when listing the bucket (with `deleteAfterRead=false` it is
consumed again once its exchange is done).
+The objects of a poll that a stopping consumer does not process are no longer
kept in the `inProgressRepository`, so
+they are consumed when the route is started again.
+
=== camel-aws2-sqs
The `batchSeparator` option (default `,`), which splits a `String` body into
the messages of the `sendBatchMessage`
