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 9b189c0637af CAMEL-25512: camel-minio - Do not consume an object again
while its exchange is in flight
9b189c0637af is described below
commit 9b189c0637afe1e7ebdbc77705725ff4a312a2e4
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 10 14:11:11 2026 +0530
CAMEL-25512: camel-minio - Do not consume an object again while its
exchange is in flight
The minio consumer routes each exchange asynchronously and deletes or
moves the object only when the exchange completes, so the next poll
listed the same object again and consumed it a second time while the
first exchange was still in flight. The consumer now keeps the names of
the objects in flight, as camel-aws2-s3 does, and skips them on the next
poll. A name is released when its exchange completes or fails, when the
batch cannot be created, and for the exchanges not routed because the
consumer is stopping.
A listed object that is deleted, for example by another consumer, before
its stat or get (NoSuchKey) is now skipped instead of failing the whole
poll, and is not counted as polled. As the batch properties are set up
front, CamelBatchSize can then be higher than the number of routed
exchanges, which the component docs note.
Closes #27645
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/minio-component.adoc | 4 +
.../camel-minio/src/main/docs/minio-component.adoc | 4 +
.../camel/component/minio/MinioConsumer.java | 76 ++++++-
.../apache/camel/component/minio/FakeS3Server.java | 219 +++++++++++++++++++++
.../minio/MinioConsumerDeletedObjectTest.java | 108 ++++++++++
.../minio/MinioConsumerInProgressFailureTest.java | 80 ++++++++
.../minio/MinioConsumerInProgressTest.java | 128 ++++++++++++
7 files changed, 610 insertions(+), 9 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/minio-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/minio-component.adoc
index 508dcf099905..a0e2de922b1a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/minio-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/minio-component.adoc
@@ -89,6 +89,10 @@ This allows you, for instance, to know how many messages
exist in this
batch and for instance, let the Aggregator
aggregate this number of messages.
+NOTE: The batch properties are set before the content of each object is
fetched. If an object is deleted (for example
+by another consumer) between the listing and the get of its content, it is
skipped, so `CamelBatchSize` can be higher
+than the number of exchanges routed for the batch, and possibly no exchange
has `CamelBatchComplete` set to `true`.
+
=== Minio Producer operations
Camel-Minio component provides the following operation on the producer side:
diff --git a/components/camel-minio/src/main/docs/minio-component.adoc
b/components/camel-minio/src/main/docs/minio-component.adoc
index 508dcf099905..a0e2de922b1a 100644
--- a/components/camel-minio/src/main/docs/minio-component.adoc
+++ b/components/camel-minio/src/main/docs/minio-component.adoc
@@ -89,6 +89,10 @@ This allows you, for instance, to know how many messages
exist in this
batch and for instance, let the Aggregator
aggregate this number of messages.
+NOTE: The batch properties are set before the content of each object is
fetched. If an object is deleted (for example
+by another consumer) between the listing and the get of its content, it is
skipped, so `CamelBatchSize` can be higher
+than the number of exchanges routed for the batch, and possibly no exchange
has `CamelBatchComplete` set to `true`.
+
=== Minio Producer operations
Camel-Minio component provides the following operation on the producer side:
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 5f87f8858986..d1bcebcb75cc 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
@@ -21,6 +21,8 @@ import java.util.Deque;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import io.minio.BucketExistsArgs;
import io.minio.CopyObjectArgs;
@@ -31,6 +33,7 @@ import io.minio.MinioClient;
import io.minio.RemoveObjectArgs;
import io.minio.Result;
import io.minio.SourceObject;
+import io.minio.errors.ErrorResponseException;
import io.minio.errors.MinioException;
import io.minio.messages.Item;
import org.apache.camel.Exchange;
@@ -60,6 +63,8 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(MinioConsumer.class);
+ // objects whose exchanges are being processed, which a later poll must
not consume again
+ private final Set<String> inProgress = ConcurrentHashMap.newKeySet();
private long totalCounter;
private String continuationToken;
private transient String minioConsumerToString;
@@ -189,8 +194,7 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
protected Deque<Exchange> createExchanges(String objectName) throws
Exception {
Deque<Exchange> answer = new LinkedList<>();
- Exchange exchange = createExchange(objectName);
- answer.add(exchange);
+ addExchange(objectName, answer, false);
return answer;
}
@@ -202,8 +206,7 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
do {
messageCounter++;
Item minioObjectSummary =
minioObjectSummaries.next().get();
- Exchange exchange =
createExchange(minioObjectSummary.objectName());
- answer.add(exchange);
+ addExchange(minioObjectSummary.objectName(), answer, true);
} while (minioObjectSummaries.hasNext());
} else {
do {
@@ -211,8 +214,7 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
Item minioObjectSummary =
minioObjectSummaries.next().get();
// ignore if directory
if (!minioObjectSummary.isDir()) {
- Exchange exchange =
createExchange(minioObjectSummary.objectName());
- answer.add(exchange);
+ addExchange(minioObjectSummary.objectName(), answer,
true);
}
} while (minioObjectSummaries.hasNext());
}
@@ -224,12 +226,47 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
} catch (Exception e) {
LOG.warn("Error getting MinioObject due: {}", e.getMessage());
+ // none of the objects of this poll is processed
+ releaseInProgress(null, answer);
throw e;
}
return answer;
}
+ private void 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;
+ }
+ try {
+ answer.add(createExchange(objectName));
+ } 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;
+ }
+ throw e;
+ }
+ }
+
+ private static boolean isNoSuchKey(Exception e) {
+ return e instanceof ErrorResponseException ere && ere.errorResponse()
!= null
+ && "NoSuchKey".equals(ere.errorResponse().code());
+ }
+
+ private void releaseInProgress(String objectName, Queue<?> notProcessed) {
+ if (objectName != null) {
+ inProgress.remove(objectName);
+ }
+ for (Object exchange : notProcessed) {
+ inProgress.remove(((Exchange)
exchange).getIn().getHeader(MinioConstants.OBJECT_NAME, String.class));
+ }
+ }
+
private InputStream getObject(String bucketName, MinioClient minioClient,
String objectName) throws Exception {
GetObjectArgs.Builder getObjectRequest =
GetObjectArgs.builder().bucket(bucketName).object(objectName);
@@ -251,6 +288,7 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
@Override
public int processBatch(Queue<Object> exchanges) throws Exception {
int total = exchanges.size();
+ int skipped = 0;
for (int index = 0; index < total && isBatchAllowed(); index++) {
// only loop if we are started (allowed to run)
@@ -279,7 +317,16 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
});
}
} catch (Exception e) {
+ if (isNoSuchKey(e)) {
+ // deleted (for example by another consumer) after it
was listed
+ LOG.debug("Skipping object {} as it no longer exists",
srcObjectName);
+ inProgress.remove(srcObjectName);
+ 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;
}
}
@@ -287,11 +334,19 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
// add on completion to handle after work when the exchange is done
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
public void onComplete(Exchange exchange) {
- processCommit(exchange);
+ try {
+ processCommit(exchange);
+ } finally {
+ inProgress.remove(srcObjectName);
+ }
}
public void onFailure(Exchange exchange) {
- processRollback(exchange);
+ try {
+ processRollback(exchange);
+ } finally {
+ inProgress.remove(srcObjectName);
+ }
}
@Override
@@ -302,8 +357,11 @@ public class MinioConsumer extends
ScheduledBatchPollingConsumer {
getAsyncProcessor().process(exchange, EmptyAsyncCallback.get());
}
+ // the remaining exchanges are not processed as the consumer is
stopping
+ releaseInProgress(null, exchanges);
- return total;
+ // 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
new file mode 100644
index 000000000000..2769bd8c973e
--- /dev/null
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/FakeS3Server.java
@@ -0,0 +1,219 @@
+/*
+ * 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.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URLDecoder;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.sun.net.httpserver.HttpExchange;
+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.
+ */
+final class FakeS3Server {
+
+ private final String bucket;
+ private final HttpServer server;
+ private final NavigableMap<String, byte[]> objects = new
ConcurrentSkipListMap<>();
+ private final Map<String, AtomicInteger> gets = new ConcurrentHashMap<>();
+ private final Set<String> deleteWhenListed = ConcurrentHashMap.newKeySet();
+ private final Set<String> deleteWhenStatted =
ConcurrentHashMap.newKeySet();
+
+ FakeS3Server(String bucket) throws IOException {
+ this.bucket = bucket;
+ this.server = HttpServer.create(new InetSocketAddress("localhost", 0),
0);
+ this.server.createContext("/", this::handle);
+ }
+
+ void start() {
+ server.start();
+ }
+
+ void stop() {
+ server.stop(0);
+ }
+
+ String endpoint() {
+ return "http://localhost:"; + server.getAddress().getPort();
+ }
+
+ void put(String key, String content) {
+ objects.put(key, content.getBytes(StandardCharsets.UTF_8));
+ }
+
+ /**
+ * The object is deleted right after a listing returned it, as if another
consumer deleted it before the stat.
+ */
+ void deleteWhenListed(String key) {
+ deleteWhenListed.add(key);
+ }
+
+ /**
+ * The object is deleted right after a stat (HEAD) of it, as if another
consumer deleted it before the get.
+ */
+ void deleteWhenStatted(String key) {
+ deleteWhenStatted.add(key);
+ }
+
+ int gets(String key) {
+ AtomicInteger n = gets.get(key);
+ return n != null ? n.get() : 0;
+ }
+
+ private void handle(HttpExchange http) throws IOException {
+ String path = http.getRequestURI().getRawPath();
+ String key = path.length() > bucket.length() + 2 ?
decode(path.substring(bucket.length() + 2)) : null;
+ String method = http.getRequestMethod();
+ if (key == null) {
+ if ("GET".equals(method) && query(http).containsKey("list-type")) {
+ list(http, query(http));
+ } else {
+ // head bucket, location, ...
+ respond(http, 200, null, null);
+ }
+ return;
+ }
+ byte[] data = objects.get(key);
+ switch (method) {
+ case "HEAD" -> {
+ if (data == null) {
+ respond(http, 404, null, null);
+ } else {
+ objectHeaders(http, data);
+ http.sendResponseHeaders(200, -1);
+ http.close();
+ if (deleteWhenStatted.remove(key)) {
+ objects.remove(key);
+ }
+ }
+ }
+ case "GET" -> {
+ if (data == null) {
+ noSuchKey(http, key);
+ } else {
+ gets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
+ objectHeaders(http, data);
+ http.sendResponseHeaders(200, data.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(data);
+ }
+ }
+ }
+ case "DELETE" -> {
+ objects.remove(key);
+ http.sendResponseHeaders(204, -1);
+ http.close();
+ }
+ default -> respond(http, 405, null, null);
+ }
+ }
+
+ private void list(HttpExchange http, Map<String, String> query) throws
IOException {
+ int maxKeys = query.containsKey("max-keys") ?
Integer.parseInt(query.get("max-keys")) : 1000;
+ String after = query.get("continuation-token");
+ if (after == null) {
+ after = query.get("start-after");
+ }
+ NavigableMap<String, byte[]> view = after != null ?
objects.tailMap(after, false) : objects;
+ StringBuilder contents = new StringBuilder();
+ int count = 0;
+ String last = null;
+ for (Map.Entry<String, byte[]> e : view.entrySet()) {
+ if (count == maxKeys) {
+ break;
+ }
+
contents.append("<Contents><Key>").append(e.getKey()).append("</Key>")
+
.append("<LastModified>2026-10-09T00:00:00.000Z</LastModified>")
+
.append("<ETag>\"0123456789abcdef0123456789abcdef\"</ETag>")
+
.append("<Size>").append(e.getValue().length).append("</Size>")
+
.append("<StorageClass>STANDARD</StorageClass></Contents>");
+ last = e.getKey();
+ count++;
+ }
+ for (String key : deleteWhenListed) {
+ if (contents.indexOf("<Key>" + key + "</Key>") >= 0 &&
deleteWhenListed.remove(key)) {
+ objects.remove(key);
+ }
+ }
+ boolean truncated = last != null && objects.higherKey(last) != null;
+ String xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
+ + "<ListBucketResult
xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\";>"
+ + "<Name>" + bucket + "</Name><Prefix></Prefix>"
+ + "<KeyCount>" + count + "</KeyCount><MaxKeys>" + maxKeys
+ "</MaxKeys>"
+ + "<IsTruncated>" + truncated + "</IsTruncated>"
+ + (truncated ? "<NextContinuationToken>" + last +
"</NextContinuationToken>" : "")
+ + contents + "</ListBucketResult>";
+ respond(http, 200, "application/xml", xml);
+ }
+
+ private static void objectHeaders(HttpExchange http, byte[] data) {
+ http.getResponseHeaders().add("ETag",
"\"0123456789abcdef0123456789abcdef\"");
+ http.getResponseHeaders().add("Last-Modified", "Fri, 09 Oct 2026
00:00:00 GMT");
+ http.getResponseHeaders().add("Content-Type", "text/plain");
+ http.getResponseHeaders().add("Content-Length",
String.valueOf(data.length));
+ }
+
+ 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>"
+ + "<BucketName>" + bucket + "</BucketName><Resource>/" +
bucket + "/" + key + "</Resource>"
+ + "<RequestId>1</RequestId><HostId>1</HostId></Error>";
+ respond(http, 404, "application/xml", xml);
+ }
+
+ private static void respond(HttpExchange http, int status, String
contentType, String body) throws IOException {
+ if (body == null) {
+ http.sendResponseHeaders(status, -1);
+ http.close();
+ return;
+ }
+ byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ http.getResponseHeaders().add("Content-Type", contentType);
+ http.sendResponseHeaders(status, bytes.length);
+ try (OutputStream os = http.getResponseBody()) {
+ os.write(bytes);
+ }
+ }
+
+ private static Map<String, String> query(HttpExchange http) {
+ Map<String, String> answer = new HashMap<>();
+ String q = http.getRequestURI().getRawQuery();
+ if (q != null) {
+ for (String part : q.split("&")) {
+ int eq = part.indexOf('=');
+ answer.put(decode(eq >= 0 ? part.substring(0, eq) : part), eq
>= 0 ? decode(part.substring(eq + 1)) : "");
+ }
+ }
+ return answer;
+ }
+
+ private static String decode(String s) {
+ return URLDecoder.decode(s, StandardCharsets.UTF_8);
+ }
+}
diff --git
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerDeletedObjectTest.java
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerDeletedObjectTest.java
new file mode 100644
index 000000000000..72579f6513e6
--- /dev/null
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerDeletedObjectTest.java
@@ -0,0 +1,108 @@
+/*
+ * 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.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;
+
+/**
+ * An object that is deleted (for example by another consumer) after it was
listed is skipped; the other objects of the
+ * same poll are still consumed and the poll does not fail. A skipped object
does not count as polled.
+ */
+class MinioConsumerDeletedObjectTest extends CamelTestSupport {
+
+ private final List<Exception> failedPolls = new CopyOnWriteArrayList<>();
+ private final List<Integer> polledCounts = new CopyOnWriteArrayList<>();
+ private final CountDownLatch nonEmptyPoll = new CountDownLatch(1);
+ private FakeS3Server s3;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ s3 = new FakeS3Server("bkt");
+ s3.put("a.txt", "deleted before the stat");
+ s3.put("b.txt", "deleted before the get");
+ s3.put("c.txt", "hello");
+ s3.deleteWhenListed("a.txt");
+ s3.deleteWhenStatted("b.txt");
+ s3.start();
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("recordPolls", new
DefaultPollingConsumerPollStrategy() {
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ polledCounts.add(polledMessages);
+ if (polledMessages > 0) {
+ nonEmptyPoll.countDown();
+ }
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ failedPolls.add(e);
+ return false;
+ }
+ });
+ 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"
+ + "&endpoint=" + s3.endpoint())
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void deletedObjectsAreSkipped() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ mock.assertIsSatisfied();
+ assertEquals(List.of(), failedPolls, "no poll should fail because an
object was deleted after the listing");
+
+ // b.txt was queued but skipped at the get: the poll counts only c.txt
(later polls find nothing)
+ assertTrue(nonEmptyPoll.await(10, TimeUnit.SECONDS), "the poll that
consumed c.txt should be committed");
+ assertEquals(1,
polledCounts.stream().mapToInt(Integer::intValue).sum(),
+ "only the routed object should count as polled: " +
polledCounts);
+ }
+}
diff --git
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerInProgressFailureTest.java
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerInProgressFailureTest.java
new file mode 100644
index 000000000000..f43f038f3656
--- /dev/null
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerInProgressFailureTest.java
@@ -0,0 +1,80 @@
+/*
+ * 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.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+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;
+
+/**
+ * An object whose exchange failed is no longer in progress, so a later poll
consumes it again.
+ */
+class MinioConsumerInProgressFailureTest extends CamelTestSupport {
+
+ private final AtomicInteger attempts = new AtomicInteger();
+ private FakeS3Server s3;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ s3 = new FakeS3Server("bkt");
+ s3.put("a.txt", "hello");
+ s3.start();
+ return super.createCamelContext();
+ }
+
+ @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&endpoint="
+ s3.endpoint())
+ .to("seda:work");
+
+ from("seda:work")
+ .process(e -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("first attempt
fails");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void failedObjectIsConsumedAgain() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ mock.assertIsSatisfied();
+ assertEquals(2, attempts.get());
+ }
+}
diff --git
a/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerInProgressTest.java
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerInProgressTest.java
new file mode 100644
index 000000000000..d15d8e21e82b
--- /dev/null
+++
b/components/camel-minio/src/test/java/org/apache/camel/component/minio/MinioConsumerInProgressTest.java
@@ -0,0 +1,128 @@
+/*
+ * 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.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.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+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.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * An object whose exchange is still being routed asynchronously must not be
consumed again by the next poll.
+ */
+class MinioConsumerInProgressTest extends CamelTestSupport {
+
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch release = new CountDownLatch(1);
+ private FakeS3Server s3;
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ s3 = new FakeS3Server("bkt");
+ s3.put("a.txt", "hello");
+ s3.start();
+ CamelContext context = super.createCamelContext();
+ // counts the polls that completed (the exchanges of the poll were
handed to the route)
+ 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) {
+ twoPolls.countDown();
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void stopServer() {
+ release.countDown();
+ if (s3 != null) {
+ s3.stop();
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:work")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void objectInFlightIsNotConsumedAgain() throws Exception {
+ assertNotConsumedAgain("");
+ }
+
+ @Test
+ void objectInFlightIsNotConsumedAgainWithObjectName() throws Exception {
+ assertNotConsumedAgain("&objectName=a.txt");
+ }
+
+ private void assertNotConsumedAgain(String options) throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("hello");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("minio://bkt?accessKey=xxx&secretKey=yyy®ion=us-east-1&delay=10&pollStrategy=#countPolls"
+ + options + "&endpoint=" + s3.endpoint())
+ .to("seda:work");
+ }
+ });
+
+ assertTrue(inRoute.await(20, TimeUnit.SECONDS), "the object was not
consumed");
+ // the second poll has completed while the first exchange is still
held in the seda route
+ assertTrue(twoPolls.await(20, TimeUnit.SECONDS), "the consumer did not
poll again");
+ int gets = s3.gets("a.txt");
+ release.countDown();
+
+ mock.assertIsSatisfied();
+ assertEquals(1, gets,
+ "a.txt was consumed again by a later poll while its first
exchange was still in flight");
+ }
+}
