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 4de40eae3f68 CAMEL-25770: camel-google-storage - skip a listed object
that cannot be got instead of failing the poll (#28132)
4de40eae3f68 is described below
commit 4de40eae3f68cb5c0fef18ddf68a05df5c2bd5f0
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 21 00:31:55 2026 +0530
CAMEL-25770: camel-google-storage - skip a listed object that cannot be got
instead of failing the poll (#28132)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../google/storage/GoogleCloudStorageConsumer.java | 74 ++++++---
.../storage/localstorage/FakeStorageRpc.java | 19 +++
.../storage/localstorage/LocalStorageHelper.java | 8 +
.../unit/ConsumerObjectThatCannotBeGotTest.java | 166 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 ++
5 files changed, 255 insertions(+), 22 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 2ecf1997355f..7c1db25cb0b9 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
@@ -18,8 +18,10 @@ package org.apache.camel.component.google.storage;
import java.io.File;
import java.util.Date;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -163,46 +165,74 @@ public class GoogleCloudStorageConsumer extends
ScheduledBatchPollingConsumer {
}
Queue<Exchange> answer = new LinkedList<>();
- try {
- for (Blob blob : blobList) {
- if (includeObject(blob)) {
- String key = blob.getBlobId().getName();
- try {
- 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
- getExceptionHandler().handleException(
- "Skipping object " + key + " in bucket " +
getConfiguration().getBucketName()
- + " as it cannot
be downloaded: " + e.getMessage(),
- e);
- }
+ // the listed objects that could not be got, other than those that no
longer exist, by name
+ Map<String, RuntimeException> failures = new LinkedHashMap<>();
+ // whether at least one listed object could be got
+ boolean anyGot = false;
+ for (Blob blob : blobList) {
+ if (includeObject(blob)) {
+ String key = blob.getBlobId().getName();
+ try {
+ anyGot |= 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
+ getExceptionHandler().handleException(
+ "Skipping object " + key + " in bucket " +
getConfiguration().getBucketName()
+ + " as it cannot be
downloaded: " + e.getMessage(),
+ e);
+ } catch (RuntimeException e) {
+ // a failure to get one listed object (access denied, a
key that cannot decrypt it) 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
+ failures.put(key, e);
}
}
- } 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;
+ }
+
+ 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 objects, no connection): the poll fails,
and the backoff of the scheduled consumer
+ // applies
+ releaseInProgress(answer);
+ 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();
+ }
+ for (Map.Entry<String, RuntimeException> failure :
failures.entrySet()) {
+ getExceptionHandler().handleException("Error getting object "
+ failure.getKey() + " from bucket "
+ +
getConfiguration().getBucketName()
+ + ". The object is
skipped for this poll.",
+ failure.getValue());
+ }
}
return answer;
}
- private void addExchange(Blob blob, String key, Queue<Exchange> answer) {
+ /**
+ * Gets a listed object and adds its exchange.
+ *
+ * @return true if the object was got and its exchange added, false if it
is skipped (an exchange is still
+ * processing it, or it no longer exists)
+ */
+ private boolean 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;
+ return false;
}
try {
answer.add(createExchange(blob, key));
+ return true;
} 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;
+ return false;
}
throw e;
}
diff --git
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/FakeStorageRpc.java
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/FakeStorageRpc.java
index c11b76c75e30..ab87a818c559 100644
---
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/FakeStorageRpc.java
+++
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/FakeStorageRpc.java
@@ -33,9 +33,11 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@@ -116,6 +118,9 @@ class FakeStorageRpc extends StorageRpcTestBase {
//Bucketname -> bucket
Map<String, Bucket> buckets = new HashMap<>();
+ // fullnames of the objects whose content cannot be read (403), as for an
object the caller has no access to
+ Set<String> refusedReads = new HashSet<>();
+
private final boolean throwIfOption;
/** @param throwIfOption if true, we throw when given any option */
@@ -129,6 +134,17 @@ class FakeStorageRpc extends StorageRpcTestBase {
metadata = new HashMap<>();
contents = new HashMap<>();
buckets = new HashMap<>();
+ refusedReads = new HashSet<>();
+ }
+
+ void refuseReads(String bucket, String name) {
+ refusedReads.add(fullname(bucket, name));
+ }
+
+ private void checkReadable(String key) {
+ if (refusedReads.contains(key)) {
+ throw new StorageException(403, "Access denied: " + key);
+ }
}
@Override
@@ -326,6 +342,7 @@ class FakeStorageRpc extends StorageRpcTestBase {
@Override
public byte[] load(StorageObject storageObject, Map<Option, ?> options)
throws StorageException {
String key = fullname(storageObject);
+ checkReadable(key);
if (!contents.containsKey(key)) {
throw new StorageException(NOT_FOUND, "File not found: " + key);
}
@@ -346,6 +363,7 @@ class FakeStorageRpc extends StorageRpcTestBase {
}
}
String key = fullname(from);
+ checkReadable(key);
if (!contents.containsKey(key)) {
throw new StorageException(NOT_FOUND, "File not found: " + key);
}
@@ -381,6 +399,7 @@ class FakeStorageRpc extends StorageRpcTestBase {
}
}
String key = fullname(from);
+ checkReadable(key);
if (!contents.containsKey(key)) {
throw new StorageException(NOT_FOUND, "File not found: " + key);
}
diff --git
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/LocalStorageHelper.java
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/LocalStorageHelper.java
index 25919320dada..a02240907452 100644
---
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/LocalStorageHelper.java
+++
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/localstorage/LocalStorageHelper.java
@@ -83,6 +83,14 @@ public final class LocalStorageHelper {
.build();
}
+ /**
+ * Every read of the content of the object of the static FakeStorageRpc
instance is refused with 403, until the next
+ * {@link #getOptions()}.
+ */
+ public static void refuseReads(String bucket, String name) {
+ INSTANCE.refuseReads(bucket, name);
+ }
+
/**
* Returns a {@link StorageOptions} that creates a new FakeStorageRpc
instance with the given option.
*/
diff --git
a/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/unit/ConsumerObjectThatCannotBeGotTest.java
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/unit/ConsumerObjectThatCannotBeGotTest.java
new file mode 100644
index 000000000000..bfb8c5e71ee4
--- /dev/null
+++
b/components/camel-google/camel-google-storage/src/test/java/org/apache/camel/component/google/storage/unit/ConsumerObjectThatCannotBeGotTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.unit;
+
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.google.cloud.storage.BlobId;
+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.GoogleCloudStorageComponent;
+import
org.apache.camel.component.google.storage.localstorage.LocalStorageHelper;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A listed object that cannot be got (access denied, an object encrypted with
a key the consumer cannot use) must not
+ * stop the consumer: the other listed objects are consumed, and the poll
fails only when none of the listed objects
+ * could be got. The consumer lists the whole bucket at every poll, so before
this one such object stopped the consumer
+ * from consuming any object of the bucket.
+ */
+class ConsumerObjectThatCannotBeGotTest extends CamelTestSupport {
+
+ private static final String BUCKET = "bkt";
+
+ private Storage storage;
+ private final CountDownLatch failedPolls = new CountDownLatch(2);
+ private final CountDownLatch skippedReported = new CountDownLatch(1);
+ private final List<String> handled = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ storage = LocalStorageHelper.getOptions().getService();
+ storage.create(BucketInfo.of(BUCKET));
+ context.getComponent("google-storage",
GoogleCloudStorageComponent.class).getConfiguration()
+ .setStorageClient(storage);
+ 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) {
+ // noop
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ failedPolls.countDown();
+ return false;
+ }
+ });
+ context.getRegistry().bind("recordErrors", new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ handled.add(String.valueOf(exception.getMessage()));
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ record(message);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ record(message);
+ }
+
+ private void record(String message) {
+ handled.add(message);
+ if (message.contains("b.txt")) {
+ skippedReported.countDown();
+ }
+ }
+ });
+ return context;
+ }
+
+ @Test
+ void objectThatCannotBeGotDoesNotBlockTheOtherObjects() throws Exception {
+ put("a.txt");
+ put("b.txt");
+ put("c.txt");
+ LocalStorageHelper.refuseReads(BUCKET, "b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a.txt", "c.txt");
+
+ addConsumerRoute("&exceptionHandler=#recordErrors");
+
+ mock.assertIsSatisfied();
+ assertTrue(skippedReported.await(20, TimeUnit.SECONDS), "the skipped
object is reported: " + handled);
+ assertNotNull(storage.get(BlobId.of(BUCKET, "b.txt")), "b.txt stays");
+ }
+
+ @Test
+ void pollFailsWhenNoListedObjectCanBeGot() throws Exception {
+ put("a.txt");
+ put("b.txt");
+ LocalStorageHelper.refuseReads(BUCKET, "a.txt");
+ LocalStorageHelper.refuseReads(BUCKET, "b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(0);
+
+
addConsumerRoute("&pollStrategy=#countPolls&exceptionHandler=#recordErrors");
+
+ // as for a failed listing (expired credentials, no permission to get
objects): the poll fails, so the error
+ // handling and the backoff of the scheduled consumer apply
+ assertTrue(failedPolls.await(20, TimeUnit.SECONDS), "the polls fail");
+ mock.assertIsSatisfied();
+ // the failed poll reaches the exception handler, and no object is
reported as skipped before it
+ assertTrue(handled.stream().anyMatch(m -> m.startsWith("Failed polling
endpoint")),
+ "the failed polls are reported: " + handled);
+ assertNull(skippedReportedMessage(), "no object is reported as
skipped: " + handled);
+ }
+
+ private String skippedReportedMessage() {
+ return handled.stream().filter(m ->
m.contains("skipped")).findFirst().orElse(null);
+ }
+
+ private void put(String name) {
+ storage.create(BlobInfo.newBuilder(BUCKET, name).build(),
name.getBytes(StandardCharsets.UTF_8));
+ }
+
+ private void addConsumerRoute(String options) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("google-storage://" + BUCKET + "?initialDelay=0&delay=10"
+ options)
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 6804586084fc..df6c80a2e330 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1837,6 +1837,16 @@ a Spring Boot or Quarkus app that `camel run` runs
(`<name>.log`), also while th
tools keep, and the history of the AI prompt. Directories are not counted, and
`--clean` counts only the files it
removed.
+=== camel-google-storage - a listed object that cannot be got is skipped
+
+When the consumer lists the bucket, a failure to download one of the listed
objects (for example access denied, or an
+object whose encryption key the consumer cannot use) no longer fails the poll,
as long as another listed object could
+be got: the object is skipped, the other listed objects are consumed, and the
skipped object is tried again by a later
+poll. The failure is given to the exception handler of the consumer, with the
object name in the message (by default a
+`WARN` with the stack trace, at every poll that lists the object). Before, the
poll failed, and as every poll lists the
+whole bucket, one such object stopped the consumer from consuming any object.
The poll still fails when none of the
+listed objects could be got.
+
=== camel-mongodb-gridfs - the consumer routes a file once
With the `TimeStamp` (default) and `PersistentTimestamp` query strategies, the
`mongodb-gridfs` consumer routed
