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 b7d85696b1ed CAMEL-25770: camel-minio - skip a listed object that
cannot be got instead of failing the poll (#28133)
b7d85696b1ed is described below
commit b7d85696b1edcec50e8575a601b32786448f213c
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 20 16:01:41 2026 +0530
CAMEL-25770: camel-minio - skip a listed object that cannot be got instead
of failing the poll (#28133)
The minio consumer stats every listed object when it creates the exchanges
and
gets each body when it routes the batch. Any failure other than a missing
object
(access denied, an object encrypted with another SSE-C key) failed the
poll, and
as the next poll lists the same objects again, one such object kept the
consumer
from consuming the rest of the bucket.
As camel-aws2-s3 does since CAMEL-25558, a failed stat or get of one listed
object is now reported to the exception handler with its name and skipped
for
this poll; the other objects are consumed. The poll still fails when none
of the
listed objects could be got, when the listing fails, or with objectName set.
The next poll now continues after the last listed object (as
camel-alibaba-oss
does), also when the poll fails, so a refused last object of a full page no
longer holds the listing back. The exchange of an object whose body cannot
be
got is given back to the exchange factory. The behaviour change is noted in
the
4.24 upgrade guide.
Closes #28133
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/component/minio/MinioConsumer.java | 140 ++++++++++++---
.../apache/camel/component/minio/FakeS3Server.java | 40 ++++-
...merObjectThatCannotBeGotPooledExchangeTest.java | 65 +++++++
.../MinioConsumerObjectThatCannotBeGotTest.java | 189 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 ++
5 files changed, 416 insertions(+), 30 deletions(-)
diff --git
a/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java
b/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java
index d1bcebcb75cc..a0e6cc460539 100644
---
a/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java
+++
b/components/camel-minio/src/main/java/org/apache/camel/component/minio/MinioConsumer.java
@@ -19,7 +19,9 @@ package org.apache.camel.component.minio;
import java.io.InputStream;
import java.util.Deque;
import java.util.Iterator;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
+import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -67,6 +69,10 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
private final Set<String> inProgress = ConcurrentHashMap.newKeySet();
private long totalCounter;
private String continuationToken;
+ // the name of the last object listed by the poll, and the number of
objects it listed (set once the listing is
+ // read to the end)
+ private String lastListedObjectName;
+ private int listedObjects;
private transient String minioConsumerToString;
public MinioConsumer(MinioEndpoint endpoint, Processor processor) {
@@ -173,11 +179,21 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
forceConsumerAsReady();
if (listObjects.hasNext()) {
- exchanges = createExchanges(listObjects);
- if (maxMessagesPerPoll <= 0 || exchanges.size() <
maxMessagesPerPoll) {
- continuationToken = null;
- } else {
- continuationToken =
exchanges.getLast().getIn().getHeader(MinioConstants.OBJECT_NAME, String.class);
+ lastListedObjectName = null;
+ listedObjects = 0;
+ try {
+ exchanges = createExchanges(listObjects);
+ } finally {
+ // the next poll continues after the last listed object,
also when objects of this poll are skipped
+ // or none of them could be got: such objects must not
hold the consumer back, and they are listed
+ // again when the listing starts over
+ if (lastListedObjectName != null) {
+ if (maxMessagesPerPoll <= 0 || listedObjects <
maxMessagesPerPoll) {
+ continuationToken = null;
+ } else {
+ continuationToken = lastListedObjectName;
+ }
+ }
}
if (LOG.isTraceEnabled()) {
LOG.trace("Found {} objects in bucket {}...",
totalCounter, bucketName);
@@ -201,23 +217,31 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
protected Deque<Exchange> createExchanges(Iterator<Result<Item>>
minioObjectSummaries) throws Exception {
int messageCounter = 0;
Deque<Exchange> answer = new LinkedList<>();
+ // the listed objects whose stat failed, other than those that no
longer exist, by object name
+ Map<String, Exception> failures = new LinkedHashMap<>();
+ // whether the stat of at least one listed object succeeded
+ boolean anyGot = false;
+ String lastListed = null;
try {
- if (getConfiguration().isIncludeFolders()) {
- do {
- messageCounter++;
- Item minioObjectSummary =
minioObjectSummaries.next().get();
- addExchange(minioObjectSummary.objectName(), answer, true);
- } while (minioObjectSummaries.hasNext());
- } else {
- do {
- messageCounter++;
- Item minioObjectSummary =
minioObjectSummaries.next().get();
- // ignore if directory
- if (!minioObjectSummary.isDir()) {
- addExchange(minioObjectSummary.objectName(), answer,
true);
+ do {
+ messageCounter++;
+ Item minioObjectSummary = minioObjectSummaries.next().get();
+ // ignore if directory, unless folders are included
+ if (getConfiguration().isIncludeFolders() ||
!minioObjectSummary.isDir()) {
+ String objectName = minioObjectSummary.objectName();
+ // 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(objectName, answer, true);
+ } catch (Exception e) {
+ failures.put(objectName, e);
}
- } while (minioObjectSummaries.hasNext());
- }
+ }
+ lastListed = minioObjectSummary.objectName();
+ } while (minioObjectSummaries.hasNext());
+ lastListedObjectName = lastListed;
+ listedObjects = messageCounter;
if (LOG.isTraceEnabled()) {
LOG.trace("Received {} messages in this poll", messageCounter);
@@ -231,28 +255,61 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
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, no connection): the poll fails, and the backoff
of the scheduled consumer applies
+ releaseInProgress(null, 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",
+ getConfiguration().getBucketName(), failures.size(),
first.getKey());
+ throw first.getValue();
+ }
+ for (Map.Entry<String, Exception> failure : failures.entrySet()) {
+ handleSkippedObject(failure.getKey(), failure.getValue());
+ }
+ }
+
return answer;
}
- private void addExchange(String objectName, Deque<Exchange> answer,
boolean listed) throws Exception {
+ /**
+ * Creates the exchange of an object and adds it.
+ *
+ * @return true if the exchange was added, false if the object is skipped
(an exchange of a previous poll is still
+ * processing it, or it no longer exists)
+ */
+ private boolean addExchange(String objectName, Deque<Exchange> answer,
boolean listed) throws Exception {
// skip the object if an exchange of a previous poll is still
processing it
if (!inProgress.add(objectName)) {
LOG.trace("Skipping object {} as it is already in progress",
objectName);
- return;
+ return false;
}
try {
answer.add(createExchange(objectName));
+ return true;
} catch (Exception e) {
inProgress.remove(objectName);
if (listed && isNoSuchKey(e)) {
// deleted (for example by another consumer) after it was
listed
LOG.debug("Skipping object {} as it no longer exists",
objectName);
- return;
+ return false;
}
throw e;
}
}
+ private void handleSkippedObject(String objectName, Exception cause) {
+ getExceptionHandler().handleException("Error getting object " +
objectName + " from bucket "
+ +
getConfiguration().getBucketName()
+ + ". The object is skipped for
this poll.",
+ cause);
+ }
+
+ private boolean objectNameConfigured() {
+ return isNotEmpty(getConfiguration().getObjectName());
+ }
+
private static boolean isNoSuchKey(Exception e) {
return e instanceof ErrorResponseException ere && ere.errorResponse()
!= null
&& "NoSuchKey".equals(ere.errorResponse().code());
@@ -289,6 +346,10 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
public int processBatch(Queue<Object> exchanges) throws Exception {
int total = exchanges.size();
int skipped = 0;
+ // the listed objects whose body could not be got, other than those
that no longer exist, by object name
+ Map<String, Exception> failures = new LinkedHashMap<>();
+ // whether the body of at least one object of the batch could be got
+ boolean anyGot = false;
for (int index = 0; index < total && isBatchAllowed(); index++) {
// only loop if we are started (allowed to run)
@@ -308,6 +369,7 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
try {
minioObject = getObject(srcBucketName, getMinioClient(),
srcObjectName);
exchange.getIn().setBody(IOUtils.toByteArray(minioObject));
+ anyGot = true;
if (getConfiguration().isAutoCloseBody()) {
exchange.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
@Override
@@ -317,6 +379,8 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
});
}
} catch (Exception e) {
+ // the exchange is not routed, so nothing else gives it
back to the exchange factory
+ releaseExchange(exchange, false);
if (isNoSuchKey(e)) {
// deleted (for example by another consumer) after it
was listed
LOG.debug("Skipping object {} as it no longer exists",
srcObjectName);
@@ -324,10 +388,18 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
skipped++;
continue;
}
- LOG.warn("Error getting MinioObject due: {}",
e.getMessage());
- // this and the remaining exchanges of the batch are not
processed
- releaseInProgress(srcObjectName, exchanges);
- throw e;
+ if (objectNameConfigured()) {
+ LOG.warn("Error getting MinioObject due: {}",
e.getMessage());
+ // the one object of the poll is not processed
+ releaseInProgress(srcObjectName, exchanges);
+ throw e;
+ }
+ // a failure to get one listed object does not stop the
other objects of the batch: the object is
+ // skipped, and a later poll tries it again
+ inProgress.remove(srcObjectName);
+ skipped++;
+ failures.put(srcObjectName, e);
+ continue;
}
}
@@ -360,6 +432,20 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
// the remaining exchanges are not processed as the consumer is
stopping
releaseInProgress(null, exchanges);
+ if (!failures.isEmpty()) {
+ if (!anyGot) {
+ // none of the objects of the batch could be got, so this is
not about one object: the poll fails, and
+ // the backoff of the scheduled consumer applies
+ Map.Entry<String, Exception> first =
failures.entrySet().iterator().next();
+ LOG.warn("None of the 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, Exception> failure : failures.entrySet()) {
+ handleSkippedObject(failure.getKey(), failure.getValue());
+ }
+ }
+
// skipped objects were not polled: they must not count for
sendEmptyMessageWhenIdle and greedy polling
return total - skipped;
}
diff --git
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/FakeS3Server.java
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/FakeS3Server.java
index 2769bd8c973e..5b5cc68d411c 100644
---
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/FakeS3Server.java
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/FakeS3Server.java
@@ -34,7 +34,8 @@ import com.sun.net.httpserver.HttpServer;
/**
* A minimal in-process fake of the S3 API calls the minio consumer makes for
a single bucket: head bucket, list objects
- * (v2, paginated with max-keys and start-after/continuation-token), stat, get
and delete object.
+ * (v2, paginated with max-keys and start-after/continuation-token), stat, get
and delete object. The gets (or the stats
+ * and gets) of an object can be refused with 403 (AccessDenied).
*/
final class FakeS3Server {
@@ -44,6 +45,8 @@ final class FakeS3Server {
private final Map<String, AtomicInteger> gets = new ConcurrentHashMap<>();
private final Set<String> deleteWhenListed = ConcurrentHashMap.newKeySet();
private final Set<String> deleteWhenStatted =
ConcurrentHashMap.newKeySet();
+ private final Set<String> refusedGets = ConcurrentHashMap.newKeySet();
+ private final Set<String> refusedStats = ConcurrentHashMap.newKeySet();
FakeS3Server(String bucket) throws IOException {
this.bucket = bucket;
@@ -81,6 +84,25 @@ final class FakeS3Server {
deleteWhenStatted.add(key);
}
+ /**
+ * Every get of the object is refused with 403 (AccessDenied), which the
client does not retry.
+ */
+ void refuseGets(String key) {
+ refusedGets.add(key);
+ }
+
+ /**
+ * Every stat and get of the object is refused with 403, as for an object
encrypted with a customer key.
+ */
+ void refuseStats(String key) {
+ refusedStats.add(key);
+ refusedGets.add(key);
+ }
+
+ boolean contains(String key) {
+ return objects.containsKey(key);
+ }
+
int gets(String key) {
AtomicInteger n = gets.get(key);
return n != null ? n.get() : 0;
@@ -102,7 +124,9 @@ final class FakeS3Server {
byte[] data = objects.get(key);
switch (method) {
case "HEAD" -> {
- if (data == null) {
+ if (refusedStats.contains(key)) {
+ respond(http, 403, null, null);
+ } else if (data == null) {
respond(http, 404, null, null);
} else {
objectHeaders(http, data);
@@ -114,7 +138,9 @@ final class FakeS3Server {
}
}
case "GET" -> {
- if (data == null) {
+ if (refusedGets.contains(key)) {
+ accessDenied(http, key);
+ } else if (data == null) {
noSuchKey(http, key);
} else {
gets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
@@ -179,6 +205,14 @@ final class FakeS3Server {
http.getResponseHeaders().add("Content-Length",
String.valueOf(data.length));
}
+ private void accessDenied(HttpExchange http, String key) throws
IOException {
+ String xml = "<?xml version=\"1.0\"
encoding=\"UTF-8\"?><Error><Code>AccessDenied</Code>"
+ + "<Message>Access Denied</Message><Key>" + key + "</Key>"
+ + "<BucketName>" + bucket + "</BucketName><Resource>/" +
bucket + "/" + key + "</Resource>"
+ + "<RequestId>1</RequestId><HostId>1</HostId></Error>";
+ respond(http, 403, "application/xml", xml);
+ }
+
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>"
diff --git
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerObjectThatCannotBeGotPooledExchangeTest.java
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerObjectThatCannotBeGotPooledExchangeTest.java
new file mode 100644
index 000000000000..7da11e43d8ad
--- /dev/null
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerObjectThatCannotBeGotPooledExchangeTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.minio;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.spi.ExchangeFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The same tests with pooled exchanges, and the exchange of an object whose
body cannot be got is given back to the
+ * pool.
+ */
+class MinioConsumerObjectThatCannotBeGotPooledExchangeTest extends
MinioConsumerObjectThatCannotBeGotTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ PooledExchangeFactory factory = new PooledExchangeFactory();
+ factory.setStatisticsEnabled(true);
+ context.getCamelContextExtension().setExchangeFactory(factory);
+ return context;
+ }
+
+ @Test
+ void exchangeOfAnObjectWhoseBodyCannotBeGotIsGivenBackToThePool() throws
Exception {
+ server.put("a.txt", "a");
+ server.put("b.txt", "b");
+ server.put("c.txt", "c");
+ server.refuseGets("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a", "c");
+
+ // polled from the test: the route is synchronous, so the exchanges
are done when poll returns
+
addConsumerRoute("&startScheduler=false&exceptionHandler=#recordErrors");
+ MinioConsumer consumer = (MinioConsumer)
context.getRoute("consumer").getConsumer();
+
context.getCamelContextExtension().getExchangeFactoryManager().setStatisticsEnabled(true);
+
+ assertEquals(2, consumer.poll());
+
+ mock.assertIsSatisfied();
+ // the exchange created for b.txt is given back, as are those of a.txt
and c.txt
+ ExchangeFactory.Statistics stats =
context.getCamelContextExtension().getExchangeFactoryManager().getStatistics();
+ assertEquals(3, stats.getCreatedCounter() +
stats.getAcquiredCounter());
+ assertEquals(3, stats.getReleasedCounter());
+ assertEquals(0, stats.getDiscardedCounter());
+ }
+}
diff --git
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerObjectThatCannotBeGotTest.java
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerObjectThatCannotBeGotTest.java
new file mode 100644
index 000000000000..d3063075119a
--- /dev/null
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerObjectThatCannotBeGotTest.java
@@ -0,0 +1,189 @@
+/*
+ * 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.minio;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+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.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A listed object that cannot be got (access denied, an object encrypted with
a customer key the consumer does not
+ * have) 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 bucket is an in-process fake of the S3
API, used through the real Minio client.
+ */
+class MinioConsumerObjectThatCannotBeGotTest extends CamelTestSupport {
+
+ FakeS3Server server;
+ private final CountDownLatch failedPolls = new CountDownLatch(2);
+ private final List<String> handled = new CopyOnWriteArrayList<>();
+ private final CountDownLatch skippedReported = new CountDownLatch(1);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ server = new FakeS3Server("bucket");
+ server.start();
+ CamelContext context = super.createCamelContext();
+ 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) {
+ handled.add(message);
+ if (message.contains("b.txt")) {
+ skippedReported.countDown();
+ }
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ handled.add(message);
+ if (message.contains("b.txt")) {
+ skippedReported.countDown();
+ }
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void stopServer() {
+ if (server != null) {
+ server.stop();
+ }
+ }
+
+ @Test
+ void objectWhoseBodyCannotBeGotDoesNotBlockTheOtherObjects() throws
Exception {
+ server.put("a.txt", "a");
+ server.put("b.txt", "b");
+ server.put("c.txt", "c");
+ server.refuseGets("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a", "c");
+
+ addConsumerRoute("&exceptionHandler=#recordErrors");
+
+ mock.assertIsSatisfied();
+ assertTrue(skippedReported.await(20, TimeUnit.SECONDS), "the skipped
object is reported: " + handled);
+ assertTrue(server.contains("b.txt"), "b.txt stays");
+ }
+
+ @Test
+ void objectThatCannotBeStatedDoesNotBlockTheOtherObjects() throws
Exception {
+ server.put("a.txt", "a");
+ server.put("b.txt", "b");
+ server.put("c.txt", "c");
+ server.refuseStats("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a", "c");
+
+ addConsumerRoute("&exceptionHandler=#recordErrors");
+
+ mock.assertIsSatisfied();
+ assertTrue(skippedReported.await(20, TimeUnit.SECONDS), "the skipped
object is reported: " + handled);
+ assertTrue(server.contains("b.txt"), "b.txt stays");
+ }
+
+ @Test
+ void pollFailsWhenNoListedObjectCanBeGot() throws Exception {
+ server.put("a.txt", "a");
+ server.put("b.txt", "b");
+ server.refuseGets("a.txt");
+ server.refuseGets("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(0);
+
+ addConsumerRoute("&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
+ assertTrue(failedPolls.await(20, TimeUnit.SECONDS), "the polls fail");
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void refusedLastListedObjectDoesNotHoldTheListingBack() throws Exception {
+ // with maxMessagesPerPoll=2 the first poll lists a.txt and b.txt (and
c.txt if the client reads the next page
+ // too): c.txt, which cannot be stated, must not keep later polls from
listing an object added before it
+ server.put("a.txt", "a");
+ server.put("b.txt", "b");
+ server.put("c.txt", "c");
+ server.refuseStats("c.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a", "b");
+
+
addConsumerRoute("&maxMessagesPerPoll=2&exceptionHandler=#recordErrors&pollStrategy=#countPolls");
+
+ mock.assertIsSatisfied();
+ // the polls that list only c.txt fail
+ assertTrue(failedPolls.await(20, TimeUnit.SECONDS), "the polls that
list only c.txt fail");
+
+ mock.reset();
+ mock.expectedBodiesReceived("a2");
+ server.put("a2.txt", "a2");
+ mock.assertIsSatisfied();
+ assertTrue(server.contains("c.txt"), "c.txt stays");
+ }
+
+ void addConsumerRoute(String options) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("minio://bucket?accessKey=xxx&secretKey=yyy®ion=us-east-1&initialDelay=0&delay=10"
+ + "&endpoint=" + server.endpoint() + options)
+ .routeId("consumer")
+ .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 6648e2c72877..6872b0344856 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
@@ -1592,6 +1592,18 @@ A started PQC data format no longer keeps a
`KeyGenerator` of its own: `getKeyGe
instance and, unless one is configured, the `CamelPQCKemAlgorithm` header now
selects the KEM algorithm of each
operation.
+=== camel-minio - a listed object that cannot be got is skipped
+
+When the consumer lists the bucket, a failure to stat or get one of the listed
objects (for example access denied, or
+an object encrypted with a customer key the consumer does not have) 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 objects listed
+after it in the same poll were not consumed, so one such object could keep the
consumer from consuming the rest of the
+bucket. The poll still fails when none of the listed objects could be got, and
when `objectName` is set. When a poll
+lists `maxMessagesPerPoll` objects, the next poll now continues after the last
listed object (before, after the last
+object that was consumed), so an object that cannot be got at the end of a
page does not hold the listing back.
+
=== camel-box - getEnterpriseEvents returns the events of every chunk
The `getEnterpriseEvents` operation of the `event-logs` API now returns all
the events between `after` and `before`.
