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 a2a592c0772b CAMEL-25760: camel-alibaba-oss - skip a listed object
that cannot be got instead of failing the poll (#28075)
a2a592c0772b is described below
commit a2a592c0772b990cfa381bc810ac0feaf224fe26
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 19 19:58:44 2026 +0530
CAMEL-25760: camel-alibaba-oss - skip a listed object that cannot be got
instead of failing the poll (#28075)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../camel/component/alibaba/oss/OSSConsumer.java | 70 +++++--
.../oss/OSSConsumerObjectThatCannotBeGotTest.java | 207 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 10 +
3 files changed, 267 insertions(+), 20 deletions(-)
diff --git
a/components/camel-alibaba/camel-alibaba-oss/src/main/java/org/apache/camel/component/alibaba/oss/OSSConsumer.java
b/components/camel-alibaba/camel-alibaba-oss/src/main/java/org/apache/camel/component/alibaba/oss/OSSConsumer.java
index e83163dcf2d0..f88f27ce06b0 100644
---
a/components/camel-alibaba/camel-alibaba-oss/src/main/java/org/apache/camel/component/alibaba/oss/OSSConsumer.java
+++
b/components/camel-alibaba/camel-alibaba-oss/src/main/java/org/apache/camel/component/alibaba/oss/OSSConsumer.java
@@ -16,8 +16,10 @@
*/
package org.apache.camel.component.alibaba.oss;
+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;
@@ -93,17 +95,16 @@ public class OSSConsumer extends
ScheduledBatchPollingConsumer {
forceConsumerAsReady();
- String nextToken = null;
+ // the next poll lists the next page, also when the objects of this
page cannot be got: such a page must not
+ // hold the consumer back, and its objects are listed again when the
listing starts over
if (Boolean.TRUE.equals(listing.isTruncated()) &&
listing.nextContinuationToken() != null) {
- nextToken = listing.nextContinuationToken();
+ continuationToken = listing.nextContinuationToken();
+ } else {
+ continuationToken = null;
}
Queue<Exchange> exchanges = createExchanges(bucketName,
listing.contents());
- int processed = processBatch(CastUtils.cast(exchanges));
- if (processed >= 0) {
- continuationToken = nextToken;
- }
- return processed;
+ return processBatch(CastUtils.cast(exchanges));
}
@Override
@@ -155,26 +156,54 @@ public class OSSConsumer extends
ScheduledBatchPollingConsumer {
return answer;
}
- try {
- for (ObjectSummary summary : summaries) {
- if (summary.key() == null || summary.key().endsWith("/")) {
- continue;
- }
- addExchange(bucketName, summary.key(), answer);
+ // the listed objects that could not be got, other than those that no
longer exist, by key
+ Map<String, Exception> failures = new LinkedHashMap<>();
+ // whether at least one listed object could be got
+ boolean anyGot = false;
+ for (ObjectSummary summary : summaries) {
+ String key = summary.key();
+ if (key == null || key.endsWith("/")) {
+ 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 {
+ anyGot |= addExchange(bucketName, key, answer);
+ } catch (Exception e) {
+ failures.put(key, 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, Exception> first =
failures.entrySet().iterator().next();
+ LOG.warn("None of the listed objects of bucket {} could be got
({} failed, the first is {}): the poll fails",
+ bucketName, failures.size(), first.getKey());
+ throw first.getValue();
+ }
+ for (Map.Entry<String, Exception> failure : failures.entrySet()) {
+ getExceptionHandler().handleException("Error getting object "
+ failure.getKey() + " from bucket "
+ + bucketName + ". The
object is skipped for this poll.",
+ failure.getValue());
}
- } catch (Exception e) {
- // none of the objects of this poll is processed
- releaseNotProcessed(answer);
- throw e;
}
return answer;
}
- private void addExchange(String bucketName, String key, Queue<Exchange>
answer) throws Exception {
+ /**
+ * 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 of a previous poll
+ * is still processing it, or it no longer exists)
+ */
+ private boolean addExchange(String bucketName, String key, Queue<Exchange>
answer) throws Exception {
// skip the object if an exchange of a previous poll is still
processing it
if (!inProgress.add(key)) {
LOG.trace("Skipping object {} as it is already in progress", key);
- return;
+ return false;
}
try {
GetObjectResult result =
ossClient.getObject(GetObjectRequest.newBuilder()
@@ -191,12 +220,13 @@ public class OSSConsumer extends
ScheduledBatchPollingConsumer {
exchange.setPattern(endpoint.getExchangePattern());
OSSUtils.mapOssObject(exchange, bucketName, key, result, body);
answer.add(exchange);
+ return true;
} catch (Exception 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-alibaba/camel-alibaba-oss/src/test/java/org/apache/camel/component/alibaba/oss/OSSConsumerObjectThatCannotBeGotTest.java
b/components/camel-alibaba/camel-alibaba-oss/src/test/java/org/apache/camel/component/alibaba/oss/OSSConsumerObjectThatCannotBeGotTest.java
new file mode 100644
index 000000000000..05ab87ecbf49
--- /dev/null
+++
b/components/camel-alibaba/camel-alibaba-oss/src/test/java/org/apache/camel/component/alibaba/oss/OSSConsumerObjectThatCannotBeGotTest.java
@@ -0,0 +1,207 @@
+/*
+ * 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.alibaba.oss;
+
+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.aliyun.sdk.service.oss2.OSSClient;
+import com.aliyun.sdk.service.oss2.exceptions.OperationException;
+import com.aliyun.sdk.service.oss2.exceptions.ServiceException;
+import com.aliyun.sdk.service.oss2.models.DeleteObjectRequest;
+import com.aliyun.sdk.service.oss2.models.GetObjectRequest;
+import com.aliyun.sdk.service.oss2.models.GetObjectResult;
+import com.aliyun.sdk.service.oss2.models.ListObjectsV2Request;
+import com.aliyun.sdk.service.oss2.models.ListObjectsV2Result;
+import com.aliyun.sdk.service.oss2.models.ObjectSummary;
+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.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * 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, the next page of the
listing is read, 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 continuation token, and answers the get of a
forbidden object with 403.
+ */
+class OSSConsumerObjectThatCannotBeGotTest extends CamelTestSupport {
+
+ 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("ossClient", ossClient());
+ 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(
+ () -> assertThat(objects).as("a.txt and c.txt are consumed and
deleted, b.txt stays").containsExactly("b.txt"));
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertThat(handled).as("the skipped object is
reported").anyMatch(m -> m.contains("b.txt")));
+ }
+
+ @Test
+ void pageOfObjectsThatCannotBeGotDoesNotBlockTheNextPage() throws
Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ objects.add("c.txt");
+ forbidden.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("b.txt", "c.txt");
+
+ // one object per page: the first page holds only a.txt
+
addConsumerRoute("&deleteAfterRead=true&maxMessagesPerPoll=1&pollStrategy=#countPolls");
+
+ mock.assertIsSatisfied();
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(
+ () -> assertThat(objects).as("b.txt and c.txt are consumed and
deleted, a.txt stays").containsExactly("a.txt"));
+ }
+
+ @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("alibaba-oss:bucket?ossClient=#ossClient&initialDelay=0&delay=10" +
options)
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+ }
+
+ private OSSClient ossClient() {
+ OSSClient client = Mockito.mock(OSSClient.class);
+
Mockito.when(client.listObjectsV2(Mockito.any(ListObjectsV2Request.class))).thenAnswer(i
-> {
+ ListObjectsV2Request request = i.getArgument(0);
+ String after = request.continuationToken();
+ long maxKeys = request.maxKeys() != null ? request.maxKeys() :
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();
+ List<ObjectSummary> contents = page.stream().map(key -> {
+ ObjectSummary summary = Mockito.mock(ObjectSummary.class);
+ Mockito.when(summary.key()).thenReturn(key);
+ return summary;
+ }).toList();
+ ListObjectsV2Result result =
Mockito.mock(ListObjectsV2Result.class);
+ Mockito.when(result.contents()).thenReturn(contents);
+ Mockito.when(result.isTruncated()).thenReturn(truncated);
+ Mockito.when(result.nextContinuationToken()).thenReturn(truncated
? page.get(page.size() - 1) : null);
+ return result;
+ });
+
Mockito.when(client.getObject(Mockito.any(GetObjectRequest.class))).thenAnswer(i
-> {
+ String key = i.<GetObjectRequest> getArgument(0).key();
+ if (!objects.contains(key)) {
+ throw new OperationException("GetObject",
ServiceException.newBuilder().statusCode(404).build());
+ }
+ if (forbidden.contains(key)) {
+ throw new OperationException("GetObject",
ServiceException.newBuilder().statusCode(403).build());
+ }
+ GetObjectResult result = Mockito.mock(GetObjectResult.class);
+ Mockito.when(result.body()).thenReturn(new
ByteArrayInputStream(key.getBytes(StandardCharsets.UTF_8)));
+ return result;
+ });
+
Mockito.when(client.deleteObject(Mockito.any(DeleteObjectRequest.class))).thenAnswer(i
-> {
+ objects.remove(i.<DeleteObjectRequest> getArgument(0).key());
+ 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 7abafb4d7275..3c04245bfe41 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
@@ -1463,6 +1463,16 @@ anything there - the JDK does not support host name
verification on the server s
client certificate with `Endpoint Identification Algorithm HTTPS is not
supported on the server side` - so a consumer
that asks for client certificates and had the option set now completes the
handshake instead of failing it.
+=== camel-alibaba-oss - 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 and the next poll listed the same
+page again, so one such object stopped the consumer from consuming any object.
The poll still fails when none of the
+listed objects could be got, and the next poll then reads the next page of the
listing.
+
=== camel-dropbox - search and get read every page of the result
The `search` operation (producer and consumer) now returns every match that
the service gives. Previously it returned
