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 d5987d709488 CAMEL-26653: camel-minio - honor maxMessagesPerPoll per
poll (#29656)
d5987d709488 is described below
commit d5987d709488491bcf7ebbd3ca5dae20e3b3b6af
Author: Pratik <[email protected]>
AuthorDate: Thu Nov 12 01:00:59 2026 -0800
CAMEL-26653: camel-minio - honor maxMessagesPerPoll per poll (#29656)
* CAMEL-26653: camel-minio - honor maxMessagesPerPoll per poll
The consumer previously used maxMessagesPerPoll only as the listing page
size, so one poll kept paging and routed every object in the bucket.
It now stops listing at maxMessagesPerPoll (one page per poll), with the
continuation token taken from what the poll actually listed.
Adds MinioConsumerMaxMessagesPerPollTest and a 4.24 upgrade-guide note.
* CAMEL-26653: Address review feedback on the maxMessagesPerPoll cap
- Use the continuation token before the configured startAfter, so a
configured startAfter no longer restarts every poll at the first page
- Do not probe hasNext() at the cap; the listing counts as not ended
- Take the continuation token from the largest object name listed
- Remove the dead listing state and note the includeVersions behaviour
- Add a regression test for startAfter with deleteAfterRead=false
---
.../camel/component/minio/MinioConsumer.java | 82 ++++++++---
.../minio/MinioConsumerMaxMessagesPerPollTest.java | 160 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 ++
3 files changed, 231 insertions(+), 23 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 a0e6cc460539..977e4473752b 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
@@ -69,10 +69,11 @@ 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)
+ // the largest name of the objects listed by the last poll (set once its
listing has been read)
private String lastListedObjectName;
- private int listedObjects;
+ // whether the listing of the last poll was read to its end (false when
the poll stopped at the limit, and
+ // left false when a listing request failed before the end of the listing
was known)
+ private boolean listingExhausted;
private transient String minioConsumerToString;
public MinioConsumer(MinioEndpoint endpoint, Processor processor) {
@@ -161,16 +162,15 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
listObjectRequest.prefix(getConfiguration().getPrefix());
}
- if (isNotEmpty(getConfiguration().getStartAfter())) {
-
listObjectRequest.startAfter(getConfiguration().getStartAfter());
- continuationToken = null;
- }
-
- // if there was a marker from previous poll then use that to
- // continue from where we left last time
+ // if there was a marker from previous poll then use that to
continue from where we left last time;
+ // the configured startAfter only marks where a pass through the
bucket begins, so it is used when
+ // there is no marker: if it took precedence on every poll, each
poll would start over at the
+ // configured value and, with a poll capped at maxMessagesPerPoll,
never get past its first page
if (isNotEmpty(continuationToken)) {
LOG.trace("Resuming from marker: {}", continuationToken);
listObjectRequest.startAfter(continuationToken);
+ } else if (isNotEmpty(getConfiguration().getStartAfter())) {
+
listObjectRequest.startAfter(getConfiguration().getStartAfter());
}
Iterator<Result<Item>> listObjects =
getMinioClient().listObjects(listObjectRequest.build()).iterator();
@@ -180,19 +180,29 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
if (listObjects.hasNext()) {
lastListedObjectName = null;
- listedObjects = 0;
+ // not known to have ended until the listing of this poll has
been read, so a listing request
+ // that fails leaves the token of the previous poll in place
+ listingExhausted = false;
+ boolean listingEnded = false;
try {
exchanges = createExchanges(listObjects);
+ listingEnded = listingExhausted;
} 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;
- }
+ // the next poll continues after the largest name this
poll listed, 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. The token is kept when the
+ // listing has not ended, and cleared when it has, so the
next poll starts a new pass through
+ // the bucket (from the configured startAfter, when one is
set). A poll that lists nothing
+ // (the object it would continue after was deleted) clears
the token in the empty listing
+ // branch below and starts over, as before.
+ // Note on includeVersions: the token is an object name,
passed to the next listing as a key
+ // marker without a version marker, so when a page ends
partway through the versions of one
+ // object, the remaining versions of that object are
listed on the next pass through the
+ // bucket, not by the next poll
+ if (listingEnded) {
+ continuationToken = null;
+ } else if (lastListedObjectName != null) {
+ continuationToken = lastListedObjectName;
}
}
if (LOG.isTraceEnabled()) {
@@ -222,6 +232,11 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
// whether the stat of at least one listed object succeeded
boolean anyGot = false;
String lastListed = null;
+ // whether the listing was read to its end: it is not when the poll
stops at the limit (the listing is
+ // not probed there, see below). Below the limit the listing is read
until the iterator has no more
+ // items, paging through pages the server capped below the limit, so
its end is only reported once the
+ // iterator is exhausted
+ boolean exhausted = false;
try {
do {
messageCounter++;
@@ -238,10 +253,31 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
failures.put(objectName, e);
}
}
- lastListed = minioObjectSummary.objectName();
- } while (minioObjectSummaries.hasNext());
+ // the token of the next poll is the largest name listed by
this poll, not simply the last item
+ // the iterator returned: the items of a page are not all in
name order (a page can end on a
+ // folder prefix that sorts before objects the page held), and
continuing after a smaller name
+ // would list objects of this page again
+ String listedName = minioObjectSummary.objectName();
+ if (lastListed == null || listedName.compareTo(lastListed) >
0) {
+ lastListed = listedName;
+ }
+ // a limit above 0 caps the objects listed by a poll, so do
not read further pages of the listing:
+ // the next poll continues after the last listed object. Do
not probe hasNext() to find out
+ // whether the listing continues at the limit: the minio
client iterator fetches the next page
+ // to answer, and that page would be discarded here and
fetched again by the next poll (the
+ // listing response itself, with its truncated flag, is not
exposed by the client). The listing
+ // is therefore treated as not ended at the limit; when it did
end at a full page, the next poll
+ // lists nothing and clears the token, which costs one empty
poll per pass through the bucket
+ if (maxMessagesPerPoll > 0 && messageCounter >=
maxMessagesPerPoll) {
+ break;
+ }
+ if (!minioObjectSummaries.hasNext()) {
+ exhausted = true;
+ break;
+ }
+ } while (true);
lastListedObjectName = lastListed;
- listedObjects = messageCounter;
+ listingExhausted = exhausted;
if (LOG.isTraceEnabled()) {
LOG.trace("Received {} messages in this poll", messageCounter);
diff --git
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerMaxMessagesPerPollTest.java
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerMaxMessagesPerPollTest.java
new file mode 100644
index 000000000000..d3e81b67b35b
--- /dev/null
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerMaxMessagesPerPollTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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.ArrayList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+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.mock.MockEndpoint;
+import org.apache.camel.support.DefaultPollingConsumerPollStrategy;
+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.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A poll lists and routes at most maxMessagesPerPoll objects: with more
objects in the bucket, the following polls
+ * continue after the last listed object until the bucket is drained, and a
later poll starts over. A configured
+ * startAfter marks where each pass through the bucket begins: the polls of a
pass still continue after the last listed
+ * object, they do not start over at startAfter on every poll.
+ */
+class MinioConsumerMaxMessagesPerPollTest extends CamelTestSupport {
+
+ private static final int OBJECTS = 30;
+ private static final int MAX_PER_POLL = 10;
+ // the objects after the configured startAfter of the second route:
name(6) to name(30)
+ private static final String START_AFTER = "o005.txt";
+ private static final int AFTER_START_AFTER = OBJECTS - 5;
+
+ private final List<Integer> polledCounts = new CopyOnWriteArrayList<>();
+ private final List<Integer> polledCountsStartAfter = new
CopyOnWriteArrayList<>();
+ private FakeS3Server s3;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ s3 = new FakeS3Server("bkt");
+ for (int i = 1; i <= OBJECTS; i++) {
+ s3.put(name(i), "content-" + i);
+ }
+ s3.start();
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("recordPolls", new
DefaultPollingConsumerPollStrategy() {
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ if (endpoint.getEndpointUri().contains("startAfter")) {
+ polledCountsStartAfter.add(polledMessages);
+ } else {
+ polledCounts.add(polledMessages);
+ }
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void stopServer() {
+ if (s3 != null) {
+ s3.stop();
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("minio://bkt?accessKey=xxx&secretKey=yyy®ion=us-east-1&delay=10&pollStrategy=#recordPolls"
+ + "&maxMessagesPerPoll=" + MAX_PER_POLL +
"&deleteAfterRead=false&endpoint=" + s3.endpoint())
+ .to("mock:result");
+
from("minio://bkt?accessKey=xxx&secretKey=yyy®ion=us-east-1&delay=10&pollStrategy=#recordPolls"
+ + "&maxMessagesPerPoll=" + MAX_PER_POLL +
"&deleteAfterRead=false&startAfter=" + START_AFTER
+ + "&endpoint=" + s3.endpoint())
+ .to("mock:resultStartAfter");
+ }
+ };
+ }
+
+ @Test
+ void pollRoutesAtMostMaxMessagesPerPoll() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ // a minimum, not an exact count: with deleteAfterRead=false the polls
keep cycling through the bucket
+ // after it has been drained, so more messages can arrive while the
assertions run
+ mock.expectedMinimumMessageCount(OBJECTS);
+ mock.assertIsSatisfied();
+
+ // the first OBJECTS messages are the whole bucket, in listing order
+ List<String> bodies = mock.getExchanges().stream()
+ .limit(OBJECTS)
+ .map(e -> e.getIn().getBody(String.class))
+ .toList();
+ List<String> expected = new ArrayList<>();
+ for (int i = 1; i <= OBJECTS; i++) {
+ expected.add("content-" + i);
+ }
+ assertEquals(expected, bodies);
+
+ // the bucket was drained by three polls of one full page each,
continuing after the last listed object,
+ // and no poll ever routed more than the limit (a poll that listed
nothing is committed with 0)
+ List<Integer> nonEmpty = polledCounts.stream().filter(n -> n >
0).toList();
+ assertTrue(nonEmpty.size() >= 3, "expected at least three non-empty
polls: " + polledCounts);
+ assertEquals(List.of(MAX_PER_POLL, MAX_PER_POLL, MAX_PER_POLL),
nonEmpty.subList(0, 3),
+ "the first polls should each route one full page: " +
polledCounts);
+ assertTrue(nonEmpty.stream().allMatch(n -> n <= MAX_PER_POLL),
+ "no poll should route more than maxMessagesPerPoll objects: "
+ polledCounts);
+ }
+
+ @Test
+ void pollWithStartAfterProgressesPastTheFirstPage() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:resultStartAfter");
+ // a minimum, not an exact count: with deleteAfterRead=false the polls
keep cycling through the objects
+ // after startAfter once a pass has covered them, so more messages can
arrive while the assertions run
+ mock.expectedMinimumMessageCount(AFTER_START_AFTER);
+ mock.assertIsSatisfied();
+
+ // the first messages after startAfter are all the objects beyond it,
in listing order: the polls of the
+ // pass continued after the last listed object instead of listing the
first page after startAfter again
+ List<String> bodies = mock.getExchanges().stream()
+ .limit(AFTER_START_AFTER)
+ .map(e -> e.getIn().getBody(String.class))
+ .toList();
+ List<String> expected = new ArrayList<>();
+ for (int i = 6; i <= OBJECTS; i++) {
+ expected.add("content-" + i);
+ }
+ assertEquals(expected, bodies);
+
+ // the pass was two full pages and a last page with the remaining
objects
+ List<Integer> nonEmpty = polledCountsStartAfter.stream().filter(n -> n
> 0).toList();
+ assertTrue(nonEmpty.size() >= 3, "expected at least three non-empty
polls: " + polledCountsStartAfter);
+ assertEquals(List.of(MAX_PER_POLL, MAX_PER_POLL, AFTER_START_AFTER - 2
* MAX_PER_POLL),
+ nonEmpty.subList(0, 3),
+ "the first polls should progress through the objects after
startAfter: " + polledCountsStartAfter);
+ assertTrue(nonEmpty.stream().allMatch(n -> n <= MAX_PER_POLL),
+ "no poll should route more than maxMessagesPerPoll objects: "
+ polledCountsStartAfter);
+ }
+
+ private static String name(int i) {
+ return String.format("o%03d.txt", i);
+ }
+}
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 a99af27d8f8c..d16e069d9cd8 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
@@ -5189,6 +5189,18 @@ bucket. The poll still fails when none of the listed
objects could be got, and w
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-minio - maxMessagesPerPoll now limits a poll to one page
+
+The consumer used `maxMessagesPerPoll` only as the page size of the bucket
listing, so one poll listed every page of
+the bucket, got the metadata of every object, and routed all of them. A poll
now lists at most `maxMessagesPerPoll`
+objects (one page, as camel-aws2-s3 does), and the next poll continues after
the last object the previous poll
+listed. Draining a large bucket therefore takes more polls than before.
Setting `maxMessagesPerPoll` to 0 or a
+negative number keeps the previous behaviour of listing the whole bucket in
one poll. Some listed objects are not
+routed (folders when `includeFolders` is false, objects still being uploaded,
objects that cannot be got), so a poll
+can route fewer objects than the limit, and the next poll continues after
them. The `startAfter` option now marks
+where each pass through the bucket begins: the polls of a pass continue after
the last listed object, and a new pass
+starts after `startAfter` again once the listing has been read to its end.
+
=== camel-minio - Server-side encryption options now apply to MinioProducer
The `serverSideEncryption` and `serverSideEncryptionCustomerKey` endpoint
options now apply to producer operations:
