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 e9d519bbd6c5 CAMEL-25760: camel-huaweicloud-obs - skip a listed object
that cannot be got instead of failing the poll (#28076)
e9d519bbd6c5 is described below
commit e9d519bbd6c5f271a30cf73556f56534edd348f0
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 19 19:58:57 2026 +0530
CAMEL-25760: camel-huaweicloud-obs - skip a listed object that cannot be
got instead of failing the poll (#28076)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../component/huaweicloud/obs/OBSConsumer.java | 63 ++++--
.../obs/OBSConsumerObjectThatCannotBeGotTest.java | 220 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++
3 files changed, 274 insertions(+), 20 deletions(-)
diff --git
a/components/camel-huawei/camel-huaweicloud-obs/src/main/java/org/apache/camel/component/huaweicloud/obs/OBSConsumer.java
b/components/camel-huawei/camel-huaweicloud-obs/src/main/java/org/apache/camel/component/huaweicloud/obs/OBSConsumer.java
index 385fb1e4f479..b287757c1100 100644
---
a/components/camel-huawei/camel-huaweicloud-obs/src/main/java/org/apache/camel/component/huaweicloud/obs/OBSConsumer.java
+++
b/components/camel-huawei/camel-huaweicloud-obs/src/main/java/org/apache/camel/component/huaweicloud/obs/OBSConsumer.java
@@ -16,8 +16,10 @@
*/
package org.apache.camel.component.huaweicloud.obs;
+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;
@@ -217,29 +219,50 @@ public class OBSConsumer extends
ScheduledBatchPollingConsumer {
*/
private Queue<Exchange> createExchanges(List<ObsObject> obsObjects) {
Queue<Exchange> answer = new LinkedList<>();
- try {
- for (ObsObject objectSummary : obsObjects) {
- String objectKey = objectSummary.getObjectKey();
- // skip the object if an exchange of a previous poll is still
processing it
- if (!inProgress.add(objectKey)) {
- LOG.trace("Skipping object {} as it is already in
progress", objectKey);
+ // the listed objects that could not be got, other than those that no
longer exist, by key
+ Map<String, RuntimeException> failures = new LinkedHashMap<>();
+ // whether at least one listed object could be got
+ boolean anyGot = false;
+ for (ObsObject objectSummary : obsObjects) {
+ String objectKey = objectSummary.getObjectKey();
+ // skip the object if an exchange of a previous poll is still
processing it
+ if (!inProgress.add(objectKey)) {
+ LOG.trace("Skipping object {} as it is already in progress",
objectKey);
+ continue;
+ }
+ // a failure to get one listed object 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
+ try {
+ addExchange(objectKey, objectSummary, answer);
+ anyGot = true;
+ } catch (ObsException e) {
+ if (e.getResponseCode() == 404) {
+ // deleted (for example by another consumer) after it was
listed
+ LOG.debug("Skipping object {} as it no longer exists",
objectKey);
continue;
}
- try {
- addExchange(objectKey, objectSummary, answer);
- } catch (ObsException e) {
- if (e.getResponseCode() == 404) {
- // deleted (for example by another consumer) after it
was listed
- LOG.debug("Skipping object {} as it no longer exists",
objectKey);
- continue;
- }
- throw e;
- }
+ failures.put(objectKey, e);
+ } catch (RuntimeException e) {
+ failures.put(objectKey, 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, no connection): the poll fails, and the
backoff of the scheduled consumer applies
+ releaseNotProcessed(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",
+ endpoint.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 "
+ +
endpoint.getBucketName()
+ + ". The object is
skipped for this poll.",
+ failure.getValue());
}
- } catch (RuntimeException e) {
- // none of the objects of this poll is processed
- releaseNotProcessed(answer);
- throw e;
}
return answer;
}
diff --git
a/components/camel-huawei/camel-huaweicloud-obs/src/test/java/org/apache/camel/component/huaweicloud/obs/OBSConsumerObjectThatCannotBeGotTest.java
b/components/camel-huawei/camel-huaweicloud-obs/src/test/java/org/apache/camel/component/huaweicloud/obs/OBSConsumerObjectThatCannotBeGotTest.java
new file mode 100644
index 000000000000..f0e97fa9076a
--- /dev/null
+++
b/components/camel-huawei/camel-huaweicloud-obs/src/test/java/org/apache/camel/component/huaweicloud/obs/OBSConsumerObjectThatCannotBeGotTest.java
@@ -0,0 +1,220 @@
+/*
+ * 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.huaweicloud.obs;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.obs.services.ObsClient;
+import com.obs.services.exception.ObsException;
+import com.obs.services.model.ListObjectsRequest;
+import com.obs.services.model.ObjectListing;
+import com.obs.services.model.ObjectMetadata;
+import com.obs.services.model.ObsObject;
+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.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 org.mockito.Mockito;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A listed object that cannot be got (access denied, an archived object that
is not restored) 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 fake bucket lists its keys in order, a page of maxKeys keys at a
time with the last key of a page as next
+ * marker, and answers the get of a forbidden object with 403.
+ */
+class OBSConsumerObjectThatCannotBeGotTest extends CamelTestSupport {
+
+ private static final String BUCKET = "bucket";
+
+ private final Set<String> objects = new ConcurrentSkipListSet<>();
+ private final Set<String> forbidden = ConcurrentHashMap.newKeySet();
+ private final AtomicInteger failedPolls = new AtomicInteger();
+ private final List<String> handled = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("obsClient", obsClient());
+ 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.incrementAndGet();
+ 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) {
+ handled.add(message);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ handled.add(message);
+ }
+ });
+ return context;
+ }
+
+ @Test
+ void objectThatCannotBeGotDoesNotBlockTheOtherObjects() throws Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ objects.add("c.txt");
+ forbidden.add("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a.txt", "c.txt");
+
+
addConsumerRoute("&deleteAfterRead=true&exceptionHandler=#recordErrors");
+
+ mock.assertIsSatisfied();
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(Set.of("b.txt"), Set.copyOf(objects),
+ "a.txt and c.txt are consumed and deleted, b.txt
stays"));
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertTrue(handled.stream().anyMatch(m ->
m.contains("b.txt")), "the skipped object is reported"));
+ }
+
+ @Test
+ void objectThatCannotBeGotDoesNotBlockTheObjectsOfItsPage() throws
Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ objects.add("c.txt");
+ objects.add("d.txt");
+ forbidden.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("b.txt", "c.txt", "d.txt");
+
+ // two objects per page: a.txt shares the first page with b.txt
+ addConsumerRoute("&deleteAfterRead=true&maxMessagesPerPoll=2");
+
+ mock.assertIsSatisfied();
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertEquals(Set.of("a.txt"), Set.copyOf(objects),
"b.txt, c.txt and d.txt are consumed and deleted"));
+ }
+
+ @Test
+ void pollFailsWhenNoListedObjectCanBeGot() throws Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ forbidden.add("a.txt");
+ forbidden.add("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(0);
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls");
+
+ // 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
+ await().atMost(20, TimeUnit.SECONDS).until(() -> failedPolls.get() >=
2);
+ mock.assertIsSatisfied();
+ }
+
+ private void addConsumerRoute(String options) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("hwcloud-obs:?bucketName=" + BUCKET +
"&obsClient=#obsClient&initialDelay=0&delay=10" + options)
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+ }
+
+ private ObsClient obsClient() {
+ ObsClient client = Mockito.mock(ObsClient.class);
+
Mockito.when(client.listObjects(Mockito.any(ListObjectsRequest.class))).thenAnswer(i
-> {
+ ListObjectsRequest request = i.getArgument(0);
+ String after = request.getMarker();
+ int maxKeys = request.getMaxKeys() > 0 ? request.getMaxKeys() :
1000;
+ List<String> remaining = objects.stream().filter(k -> after ==
null || k.compareTo(after) > 0).toList();
+ List<String> page = remaining.stream().limit(maxKeys).toList();
+ boolean truncated = remaining.size() > page.size();
+ // listed objects carry no content type, so the consumer gets each
of them
+ List<ObsObject> listed = page.stream().map(key -> {
+ ObsObject summary = new ObsObject();
+ summary.setBucketName(BUCKET);
+ summary.setObjectKey(key);
+ return summary;
+ }).toList();
+ ObjectListing listing = Mockito.mock(ObjectListing.class);
+ Mockito.when(listing.getObjects()).thenReturn(listed);
+ Mockito.when(listing.isTruncated()).thenReturn(truncated);
+ Mockito.when(listing.getNextMarker()).thenReturn(truncated ?
page.get(page.size() - 1) : null);
+ return listing;
+ });
+ Mockito.when(client.getObject(Mockito.eq(BUCKET),
Mockito.anyString())).thenAnswer(i -> {
+ String key = i.getArgument(1);
+ if (!objects.contains(key)) {
+ ObsException e = new ObsException("NoSuchKey");
+ e.setResponseCode(404);
+ throw e;
+ }
+ if (forbidden.contains(key)) {
+ ObsException e = new ObsException("AccessDenied");
+ e.setResponseCode(403);
+ throw e;
+ }
+ ObsObject object = new ObsObject();
+ object.setBucketName(BUCKET);
+ object.setObjectKey(key);
+ ObjectMetadata metadata = new ObjectMetadata();
+ metadata.setContentType("text/plain");
+ object.setMetadata(metadata);
+ object.setObjectContent(new
ByteArrayInputStream(key.getBytes(StandardCharsets.UTF_8)));
+ return object;
+ });
+ Mockito.when(client.deleteObject(Mockito.eq(BUCKET),
Mockito.anyString())).thenAnswer(i -> {
+ objects.remove(i.<String> getArgument(1));
+ return null;
+ });
+ return client;
+ }
+}
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 3c04245bfe41..4d4f8a860fcc 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
@@ -1387,6 +1387,17 @@ became `café` in the headers and the body map. A
charset that the request or
that sends ISO-8859-1 form data without declaring a charset now gets U+FFFD
replacement characters instead of its
text, and must declare its charset (for example
`application/x-www-form-urlencoded; charset=ISO-8859-1`).
+=== camel-huaweicloud-obs - a listed object that cannot be got is skipped
+
+When the consumer lists the bucket, a failure to get one of the listed objects
(for example access denied, or an
+archived object that is not restored) 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, and the
skipped object is tried again by a later
+poll. The failure is given to the exception handler of the consumer, with the
key in the message (by default a `WARN`
+with the stack trace, at every poll that lists the object). Before, the poll
failed, so the other objects listed with
+it were not consumed either, and when the bucket fitted in one listing
(`maxMessagesPerPoll`, default 10) the consumer
+consumed no object at all. The poll still fails when none of the listed
objects could be got. With the `fileName`
+option, a failed get still fails the poll.
+
=== camel-github2 - consumers and items of a poll that were not routed
The `github2` consumers (`commit`, `event`, `pullRequest`,
`pullRequestComment`, `tag`) remembered every item of a
