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 940e7ecf558f CAMEL-25519: camel-google-storage - Do not consume an
object again while its exchange is in flight (#27670)
940e7ecf558f is described below
commit 940e7ecf558f3630d87ce94936c3850bbbbad6e2
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 11 12:32:18 2026 +0530
CAMEL-25519: camel-google-storage - Do not consume an object again while
its exchange is in flight (#27670)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../google/storage/GoogleCloudStorageConsumer.java | 92 +++++++--
...oogleCloudStorageConsumerDeletedObjectTest.java | 117 +++++++++++
...eCloudStorageConsumerInProgressFailureTest.java | 79 ++++++++
.../GoogleCloudStorageConsumerInProgressTest.java | 153 +++++++++++++++
...oogleCloudStorageConsumerListingWindowTest.java | 214 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 6 +
6 files changed, 646 insertions(+), 15 deletions(-)
diff --git
a/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java
b/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java
index 8a5ed9f40708..2ecf1997355f 100644
---
a/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java
+++
b/components/camel-google/camel-google-storage/src/main/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumer.java
@@ -21,6 +21,8 @@ import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import com.google.api.gax.paging.Page;
import com.google.cloud.storage.Blob;
@@ -29,6 +31,7 @@ import com.google.cloud.storage.Bucket;
import com.google.cloud.storage.CopyWriter;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.Storage.CopyRequest;
+import com.google.cloud.storage.StorageException;
import org.apache.camel.CamelContext;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
@@ -52,6 +55,11 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(GoogleCloudStorageConsumer.class);
private final Language language;
+ // objects whose exchanges are being processed, which a later poll must
not consume again
+ private final Set<String> inProgress = ConcurrentHashMap.newKeySet();
+ // the objects that were in progress when the current poll started, before
it listed or got the objects. Only the
+ // thread that polls reads and writes it.
+ private Set<String> inProgressAtPollStart = Set.of();
public GoogleCloudStorageConsumer(GoogleCloudStorageEndpoint endpoint,
Processor processor) {
super(endpoint, processor);
@@ -89,6 +97,11 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
Queue<Exchange> exchanges;
String filter = getConfiguration().getFilter();
+ // taken before the objects are listed (or got): an object that is in
flight now can complete, and be deleted
+ // and released, after it was listed and before createExchanges looks
at it, and must then not be consumed
+ // again from the stale listing
+ inProgressAtPollStart = Set.copyOf(inProgress);
+
if (fileName != null) {
LOG.trace("Getting object in bucket [{}] with file name [{}]...",
bucketName, fileName);
@@ -140,8 +153,7 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
protected Queue<Exchange> createExchanges(Blob blob, String key) {
Queue<Exchange> answer = new LinkedList<>();
- Exchange exchange = createExchange(blob, key);
- answer.add(exchange);
+ addExchange(blob, key, answer);
return answer;
}
@@ -156,7 +168,7 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
if (includeObject(blob)) {
String key = blob.getBlobId().getName();
try {
- answer.add(createExchange(blob, key));
+ addExchange(blob, key, answer);
} catch (IllegalArgumentException e) {
// the object name is rejected as a local download
path: skip this object only, so a single
// object with such a name does not stop the rest of
the bucket from being consumed
@@ -169,12 +181,47 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
}
} catch (Exception e) {
LOG.warn("Error getting object due: {}", e.getMessage(), e);
+ // none of the objects of this poll is processed
+ releaseInProgress(answer);
throw e;
}
return answer;
}
+ private void addExchange(Blob blob, String key, Queue<Exchange> answer) {
+ // skip the object if an exchange of a previous poll is still
processing it, or was when this poll started
+ if (inProgressAtPollStart.contains(key) || !inProgress.add(key)) {
+ LOG.trace("Skipping object {} as it is already in progress", key);
+ return;
+ }
+ try {
+ answer.add(createExchange(blob, key));
+ } catch (RuntimeException e) {
+ inProgress.remove(key);
+ if (isNotFound(e)) {
+ // deleted (for example by another consumer) after it was
listed
+ LOG.debug("Skipping object {} as it no longer exists", key);
+ return;
+ }
+ throw e;
+ }
+ }
+
+ private static boolean isNotFound(Exception e) {
+ StorageException se =
ObjectHelper.getException(StorageException.class, e);
+ return se != null && se.getCode() == 404;
+ }
+
+ private void releaseInProgress(Queue<?> notProcessed) {
+ for (Object notRouted : notProcessed) {
+ Exchange exchange = (Exchange) notRouted;
+
inProgress.remove(exchange.getIn().getHeader(GoogleCloudStorageConstants.OBJECT_NAME,
String.class));
+ // the exchange is never routed, so nothing else releases it to
the exchange factory
+ releaseExchange(exchange, false);
+ }
+ }
+
/**
* Decide whether to include the Objects in the results
*
@@ -208,14 +255,23 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
// update pending number of exchanges
pendingExchanges = total - index - 1;
+ final String key =
exchange.getIn().getHeader(GoogleCloudStorageConstants.OBJECT_NAME,
String.class);
// add on completion to handle after work when the exchange is done
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
public void onComplete(Exchange exchange) {
- processCommit(exchange);
+ try {
+ processCommit(exchange);
+ } finally {
+ inProgress.remove(key);
+ }
}
public void onFailure(Exchange exchange) {
- processRollback(exchange);
+ try {
+ processRollback(exchange);
+ } finally {
+ inProgress.remove(key);
+ }
}
@Override
@@ -228,6 +284,8 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, EmptyAsyncCallback.get());
}
+ // the remaining exchanges are not processed as the consumer is
stopping
+ releaseInProgress(exchanges);
return total;
}
@@ -308,6 +366,18 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
}
Exchange exchange = createExchange(true);
+ try {
+ populateExchange(exchange, pattern, blob, key);
+ } catch (RuntimeException e) {
+ // the object is not consumed (its name is rejected as a local
path, or its download failed, for example
+ // as it no longer exists), so the exchange created for it is not
routed either
+ releaseExchange(exchange, false);
+ throw e;
+ }
+ return exchange;
+ }
+
+ private void populateExchange(Exchange exchange, ExchangePattern pattern,
Blob blob, String key) {
exchange.setPattern(pattern);
Message message = exchange.getIn();
@@ -315,14 +385,8 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
// download as file
if (getConfiguration().getDownloadFileName() != null) {
// create a dummy exchange as Exchange is needed for
expression evaluation
- String result;
- try {
- result = evaluateFileExpression(exchange,
getConfiguration().getDownloadFileName(), blob.getName());
- } catch (IllegalArgumentException e) {
- // the object is not consumed, so the exchange created for
it is not routed either
- releaseExchange(exchange, false);
- throw e;
- }
+ String result
+ = evaluateFileExpression(exchange,
getConfiguration().getDownloadFileName(), blob.getName());
if (result != null) {
File file = new File(result);
blob.downloadTo(file.toPath());
@@ -367,8 +431,6 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
message.setHeader(GoogleCloudStorageConstants.METADATA_STORAGE_CLASS,
blob.getStorageClass());
message.setHeader(GoogleCloudStorageConstants.METADATA_CREATE_TIME,
blob.getCreateTime());
message.setHeader(GoogleCloudStorageConstants.METADATA_LAST_UPDATE,
new Date(blob.getUpdateTime()));
-
- return exchange;
}
protected String evaluateFileExpression(Exchange exchange, String
downloadFileName, String blogName) {
diff --git
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerDeletedObjectTest.java
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerDeletedObjectTest.java
new file mode 100644
index 000000000000..a84cd42af13f
--- /dev/null
+++
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerDeletedObjectTest.java
@@ -0,0 +1,117 @@
+/*
+ * 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.google.storage;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.ExchangePropertyKey;
+import org.apache.camel.builder.RouteBuilder;
+import
org.apache.camel.component.google.storage.localstorage.LocalStorageHelper;
+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.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * An object deleted (for example by another consumer) after the listing and
before its content is downloaded is
+ * skipped, and the other objects of the poll are consumed.
+ */
+class GoogleCloudStorageConsumerDeletedObjectTest extends CamelTestSupport {
+
+ private final AtomicInteger failedPolls = new AtomicInteger();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ Storage delegate = LocalStorageHelper.getOptions().getService();
+ delegate.create(BucketInfo.of("myCamelBucket"));
+ for (String name : new String[] { "a.txt", "b.txt", "c.txt" }) {
+ delegate.create(BlobInfo.newBuilder("myCamelBucket",
name).build(), name.getBytes(StandardCharsets.UTF_8));
+ }
+ AtomicBoolean deleted = new AtomicBoolean();
+ // the listed blobs download their content through the delegate, in
which b.txt no longer exists
+ Storage storage = (Storage)
Proxy.newProxyInstance(Storage.class.getClassLoader(), new Class<?>[] {
Storage.class },
+ (proxy, method, args) -> {
+ Object result;
+ try {
+ result = method.invoke(delegate, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ if ("list".equals(method.getName()) &&
deleted.compareAndSet(false, true)) {
+ delegate.delete("myCamelBucket", "b.txt");
+ }
+ return result;
+ });
+ context.getComponent("google-storage",
GoogleCloudStorageComponent.class).getConfiguration()
+ .setStorageClient(storage);
+ context.getRegistry().bind("countFailedPolls", new
PollingConsumerPollStrategy() {
+ @Override
+ public boolean begin(Consumer consumer, Endpoint endpoint) {
+ return true;
+ }
+
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ // only failed polls are counted
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ failedPolls.incrementAndGet();
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("google-storage://myCamelBucket?delay=10&pollStrategy=#countFailedPolls")
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void deletedObjectIsSkipped() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a.txt", "c.txt");
+ // the batch is built after the downloads, so it only counts the
objects that are consumed
+
mock.allMessages().exchangeProperty(ExchangePropertyKey.BATCH_SIZE.getName()).isEqualTo(2);
+
+ mock.assertIsSatisfied();
+ assertThat(failedPolls)
+ .as("the poll failed because b.txt was deleted between the
listing and the download")
+ .hasValue(0);
+ }
+}
diff --git
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerInProgressFailureTest.java
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerInProgressFailureTest.java
new file mode 100644
index 000000000000..7384e965db53
--- /dev/null
+++
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerInProgressFailureTest.java
@@ -0,0 +1,79 @@
+/*
+ * 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.google.storage;
+
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import
org.apache.camel.component.google.storage.localstorage.LocalStorageHelper;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * An object whose exchange failed is no longer in progress, so a later poll
consumes it again.
+ */
+class GoogleCloudStorageConsumerInProgressFailureTest extends CamelTestSupport
{
+
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ Storage storage = LocalStorageHelper.getOptions().getService();
+ storage.create(BucketInfo.of("myCamelBucket"));
+ storage.create(BlobInfo.newBuilder("myCamelBucket", "a.txt").build(),
"hello".getBytes(StandardCharsets.UTF_8));
+ context.getComponent("google-storage",
GoogleCloudStorageComponent.class).getConfiguration()
+ .setStorageClient(storage);
+ return context;
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("google-storage://myCamelBucket?delay=10")
+ .to("seda:work");
+
+ from("seda:work")
+ .process(e -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("first attempt
fails");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void failedObjectIsConsumedAgain() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ mock.assertIsSatisfied();
+ assertThat(attempts).hasValue(2);
+ }
+}
diff --git
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerInProgressTest.java
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerInProgressTest.java
new file mode 100644
index 000000000000..b843973cf59a
--- /dev/null
+++
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerInProgressTest.java
@@ -0,0 +1,153 @@
+/*
+ * 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.google.storage;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.builder.RouteBuilder;
+import
org.apache.camel.component.google.storage.localstorage.LocalStorageHelper;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.camel.util.CastUtils;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * An object whose exchange is still being routed asynchronously must not be
consumed again by the next poll, and an
+ * object whose exchange a stopping consumer did not process is consumed again.
+ */
+class GoogleCloudStorageConsumerInProgressTest extends CamelTestSupport {
+
+ private final AtomicInteger consumed = new AtomicInteger();
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ Storage storage = LocalStorageHelper.getOptions().getService();
+ storage.create(BucketInfo.of("myCamelBucket"));
+ storage.create(BlobInfo.newBuilder("myCamelBucket", "a.txt").build(),
"hello".getBytes(StandardCharsets.UTF_8));
+ context.getComponent("google-storage",
GoogleCloudStorageComponent.class).getConfiguration()
+ .setStorageClient(storage);
+ // 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) {
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void releaseRoute() {
+ release.countDown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:work")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void objectInFlightIsNotConsumedAgain() throws Exception {
+ assertNotConsumedAgain("");
+ }
+
+ @Test
+ void objectInFlightIsNotConsumedAgainWithObjectName() throws Exception {
+ assertNotConsumedAgain("&objectName=a.txt");
+ }
+
+ @Test
+ void notProcessedObjectIsConsumedAgain() throws Exception {
+ GoogleCloudStorageEndpoint endpoint
+ = context.getEndpoint("google-storage://myCamelBucket",
GoogleCloudStorageEndpoint.class);
+ // never started, so the consumer does not process any exchange of the
batch
+ GoogleCloudStorageConsumer consumer = (GoogleCloudStorageConsumer)
endpoint.createConsumer(exchange -> {
+ });
+ List<Blob> blobs =
List.of(endpoint.getStorageClient().get("myCamelBucket", "a.txt"));
+
+ consumer.processBatch(CastUtils.cast(consumer.createExchanges(blobs)));
+
+ assertThat(consumer.createExchanges(blobs))
+ .as("a.txt stayed in progress although its exchange was not
processed")
+ .hasSize(1);
+ }
+
+ private void assertNotConsumedAgain(String options) throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("google-storage://myCamelBucket?delay=10&pollStrategy=#countPolls" +
options)
+ .process(e -> consumed.incrementAndGet())
+ .to("seda:work");
+ }
+ });
+
+ assertThat(inRoute.await(20, TimeUnit.SECONDS)).as("the object was not
consumed").isTrue();
+ // the second poll has completed while the first exchange is still
held in the seda route
+ assertThat(twoPolls.await(20, TimeUnit.SECONDS)).as("the consumer did
not poll again").isTrue();
+ int consumedWhileInFlight = consumed.get();
+ release.countDown();
+
+ assertThat(consumedWhileInFlight)
+ .as("a.txt was consumed again by a later poll while its first
exchange was still in flight")
+ .isEqualTo(1);
+ mock.assertIsSatisfied();
+ }
+}
diff --git
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerListingWindowTest.java
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerListingWindowTest.java
new file mode 100644
index 000000000000..0c2a00da4dca
--- /dev/null
+++
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/GoogleCloudStorageConsumerListingWindowTest.java
@@ -0,0 +1,214 @@
+/*
+ * 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.google.storage;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Proxy;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import
org.apache.camel.component.google.storage.localstorage.LocalStorageHelper;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.CamelEvent;
+import org.apache.camel.spi.CamelEvent.ExchangeCompletedEvent;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+import org.apache.camel.support.EventNotifierSupport;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * An object that is in flight when a poll lists (or gets) it, and whose
exchange completes (the object is deleted and
+ * no longer in progress) before the poll looks at the listed object, must not
be consumed again from the stale listing.
+ * With includeBody=false nothing is downloaded, so there is no 404 either
that would skip it.
+ */
+class GoogleCloudStorageConsumerListingWindowTest extends CamelTestSupport {
+
+ private final AtomicInteger consumed = new AtomicInteger();
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private final CountDownLatch completed = new CountDownLatch(1);
+ private final CountDownLatch windowPollDone = new CountDownLatch(1);
+ // runs once, after the next listing (or get) has read the bucket and
before the consumer receives the result
+ private final AtomicReference<Runnable> afterNextRead = new
AtomicReference<>();
+ private volatile boolean windowOpened;
+ private volatile String heldExchangeId;
+ private Storage delegate;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ delegate = LocalStorageHelper.getOptions().getService();
+ delegate.create(BucketInfo.of("myCamelBucket"));
+ delegate.create(BlobInfo.newBuilder("myCamelBucket", "a.txt").build(),
"hello".getBytes(StandardCharsets.UTF_8));
+ Storage storage = (Storage)
Proxy.newProxyInstance(Storage.class.getClassLoader(), new Class<?>[] {
Storage.class },
+ (proxy, method, args) -> {
+ Object result;
+ try {
+ result = method.invoke(delegate, args);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ boolean list = "list".equals(method.getName());
+ boolean getObject = "get".equals(method.getName()) &&
args.length > 1 && args[1] instanceof String;
+ if (list || getObject) {
+ Runnable hook = afterNextRead.getAndSet(null);
+ if (hook != null) {
+ hook.run();
+ }
+ }
+ return result;
+ });
+ context.getComponent("google-storage",
GoogleCloudStorageComponent.class).getConfiguration()
+ .setStorageClient(storage);
+ // the completed event of an exchange is sent after all its on
completions have run
+ context.getManagementStrategy().addEventNotifier(new
EventNotifierSupport() {
+ @Override
+ public void notify(CamelEvent event) {
+ if (event instanceof ExchangeCompletedEvent done
+ &&
done.getExchange().getExchangeId().equals(heldExchangeId)) {
+ completed.countDown();
+ }
+ }
+ });
+ 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) {
+ if (windowOpened) {
+ windowPollDone.countDown();
+ }
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void releaseRoute() {
+ release.countDown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions of the consumer (the delete,
then the release of the object name) over
+ // to this route
+ from("seda:work")
+ .process(e -> {
+ heldExchangeId = e.getExchangeId();
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void objectCompletedAfterTheListingIsNotConsumedAgain() throws Exception {
+ assertNotConsumedFromStaleRead("");
+ }
+
+ @Test
+ void objectCompletedAfterTheGetIsNotConsumedAgainWithObjectName() throws
Exception {
+ assertNotConsumedFromStaleRead("&objectName=a.txt");
+ }
+
+ @Test
+ void exchangeOfAnObjectThatNoLongerExistsIsReleased() throws Exception {
+ GoogleCloudStorageEndpoint endpoint
+ = context.getEndpoint("google-storage://myCamelBucket",
GoogleCloudStorageEndpoint.class);
+ List<Exchange> released = new ArrayList<>();
+ GoogleCloudStorageConsumer consumer = new
GoogleCloudStorageConsumer(endpoint, exchange -> {
+ }) {
+ @Override
+ public void releaseExchange(Exchange exchange, boolean
autoRelease) {
+ released.add(exchange);
+ super.releaseExchange(exchange, autoRelease);
+ }
+ };
+ Blob listed = delegate.get("myCamelBucket", "a.txt");
+ delegate.delete("myCamelBucket", "a.txt");
+
+ assertThat(consumer.createExchanges(List.of(listed))).isEmpty();
+
+ assertThat(released).as("the exchange created for a.txt, whose
download failed, was not released").hasSize(1);
+ }
+
+ private void assertNotConsumedFromStaleRead(String options) throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("google-storage://myCamelBucket?includeBody=false&delay=10&pollStrategy=#countPolls"
+ options)
+ .process(e -> consumed.incrementAndGet())
+ .to("seda:work");
+ }
+ });
+
+ assertThat(inRoute.await(20, TimeUnit.SECONDS)).as("the object was not
consumed").isTrue();
+ AtomicReference<Boolean> completedInWindow = new AtomicReference<>();
+ // the next poll reads a.txt while its exchange is in flight; before
the consumer looks at what it read, the
+ // exchange completes: a.txt is deleted and no longer in progress
+ afterNextRead.set(() -> {
+ release.countDown();
+ try {
+ completedInWindow.set(completed.await(20, TimeUnit.SECONDS));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ windowOpened = true;
+ });
+
+ assertThat(windowPollDone.await(20, TimeUnit.SECONDS)).as("the poll
that read the stale a.txt did not end").isTrue();
+ assertThat(completedInWindow.get()).as("the exchange of a.txt did not
complete during the poll").isTrue();
+ assertThat(delegate.get("myCamelBucket", "a.txt")).as("a.txt was not
deleted").isNull();
+ assertThat(consumed)
+ .as("a.txt was consumed again from a listing made while its
exchange was in flight")
+ .hasValue(1);
+ mock.assertIsSatisfied();
+ }
+}
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 ff9c0d0a3386..f1dc62dde5dc 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
@@ -3737,6 +3737,12 @@ When `downloadFileName` is fully dynamic and has no
static directory prefix (for
`downloadFileName=${file:name}`), a relative destination must stay within the
working directory. An absolute
destination that comes from the route's own expression is not confined.
+The consumer no longer consumes an object again while an exchange of an
earlier poll is still processing it, which
+happened with an asynchronous route (for example `to("seda:...")`), so the
next polls now skip such an object (with
+`deleteAfterRead=false` it is consumed again once its exchange is done). An
object that no longer exists when its
+content is downloaded (for example deleted by another consumer after the
listing) is now skipped instead of failing the
+poll.
+
=== camel-jt400
The endpoint syntax in the component metadata is now
`jt400:userID:password@systemName/objectPath`, which is what the
