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 ab5932958d36 CAMEL-25551: camel-azure-storage-blob,
camel-huaweicloud-obs, camel-alibaba-oss, camel-aws2-s3-vectors - do not
consume an object again while its exchange is in flight (#27682)
ab5932958d36 is described below
commit ab5932958d36cc4e5045a4810afeaf7281587d45
Author: Claus Ibsen <[email protected]>
AuthorDate: Sun Oct 11 09:01:12 2026 +0200
CAMEL-25551: camel-azure-storage-blob, camel-huaweicloud-obs,
camel-alibaba-oss, camel-aws2-s3-vectors - do not consume an object again while
its exchange is in flight (#27682)
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
---
.../catalog/docs/azure-storage-blob-consumer.adoc | 2 +
.../camel/component/alibaba/oss/OSSConsumer.java | 73 ++++-
.../alibaba/oss/OSSConsumerInProgressTest.java | 280 +++++++++++++++++++
.../aws2/s3vectors/AWS2S3VectorsConsumer.java | 43 ++-
.../AWS2S3VectorsConsumerInProgressTest.java | 163 +++++++++++
.../src/main/docs/azure-storage-blob-consumer.adoc | 2 +
.../component/azure/storage/blob/BlobConsumer.java | 82 +++++-
.../storage/blob/BlobConsumerInProgressTest.java | 277 +++++++++++++++++++
.../component/huaweicloud/obs/OBSConsumer.java | 92 ++++++-
.../huaweicloud/obs/OBSConsumerInProgressTest.java | 300 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 20 ++
11 files changed, 1299 insertions(+), 35 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-consumer.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-consumer.adoc
index 278efa914273..a15d2b697f50 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-consumer.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-blob-consumer.adoc
@@ -171,6 +171,8 @@ IMPORTANT: The delete operation is only performed if the
Exchange is successfull
NOTE: When `deleteAfterRead` is set to `false` (the default), the same blobs
will be retrieved repeatedly in subsequent polls. In this case, you should use
the Idempotent Consumer EIP to filter out duplicates based on the
`CamelAzureStorageBlobBlobName` header.
+NOTE: A blob is not retrieved again while the exchange of a previous poll is
still processing it, for example when the route continues asynchronously (such
as `to("seda:...")`). It is retrieved again by a later poll once that exchange
is done, unless it was deleted or moved. A blob that was deleted (for example
by another consumer) between the listing and its download is skipped.
+
=== Move After Read
The consumer can move blobs to a different container after successful
processing. This is useful for archiving processed blobs or implementing a
processing pipeline across containers.
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 4c285604ead4..e83163dcf2d0 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
@@ -19,8 +19,11 @@ package org.apache.camel.component.alibaba.oss;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import com.aliyun.sdk.service.oss2.OSSClient;
+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;
@@ -44,6 +47,8 @@ public class OSSConsumer extends
ScheduledBatchPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(OSSConsumer.class);
private final OSSEndpoint endpoint;
+ // objects whose exchanges are being processed, which a later poll must
not consume again
+ private final Set<String> inProgress = ConcurrentHashMap.newKeySet();
private OSSClient ossClient;
private String continuationToken;
@@ -114,21 +119,32 @@ public class OSSConsumer extends
ScheduledBatchPollingConsumer {
pendingExchanges = total - index - 1;
+ final String objectKey =
exchange.getIn().getHeader(OSSHeaders.OBJECT_KEY, String.class);
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
@Override
public void onComplete(Exchange exchange) {
- processComplete(exchange);
+ try {
+ processComplete(exchange);
+ } finally {
+ inProgress.remove(objectKey);
+ }
}
@Override
public void onFailure(Exchange exchange) {
- processFailure(exchange);
+ try {
+ processFailure(exchange);
+ } finally {
+ inProgress.remove(objectKey);
+ }
}
});
AsyncCallback callback = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, callback);
}
+ // the remaining exchanges are not processed as the consumer is
stopping
+ releaseNotProcessed(exchanges);
return total;
}
@@ -139,14 +155,31 @@ public class OSSConsumer extends
ScheduledBatchPollingConsumer {
return answer;
}
- for (ObjectSummary summary : summaries) {
- if (summary.key() == null || summary.key().endsWith("/")) {
- continue;
+ try {
+ for (ObjectSummary summary : summaries) {
+ if (summary.key() == null || summary.key().endsWith("/")) {
+ continue;
+ }
+ addExchange(bucketName, summary.key(), answer);
}
+ } 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 {
+ // 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;
+ }
+ try {
GetObjectResult result =
ossClient.getObject(GetObjectRequest.newBuilder()
.bucket(bucketName)
- .key(summary.key())
+ .key(key)
.build());
byte[] body;
@@ -156,10 +189,34 @@ public class OSSConsumer extends
ScheduledBatchPollingConsumer {
Exchange exchange = createExchange(true);
exchange.setPattern(endpoint.getExchangePattern());
- OSSUtils.mapOssObject(exchange, bucketName, summary.key(), result,
body);
+ OSSUtils.mapOssObject(exchange, bucketName, key, result, body);
answer.add(exchange);
+ } 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;
+ }
+ throw e;
}
- return answer;
+ }
+
+ private static boolean isNotFound(Exception e) {
+ ServiceException se =
ObjectHelper.getException(ServiceException.class, e);
+ return se != null && se.statusCode() == 404;
+ }
+
+ private void releaseNotProcessed(Queue<?> exchanges) {
+ for (Object o : exchanges) {
+ Exchange exchange = (Exchange) o;
+ String key = exchange.getIn().getHeader(OSSHeaders.OBJECT_KEY,
String.class);
+ if (key != null) {
+ inProgress.remove(key);
+ }
+ releaseExchange(exchange, true);
+ }
+ exchanges.clear();
}
private void processComplete(Exchange exchange) {
diff --git
a/components/camel-alibaba/camel-alibaba-oss/src/test/java/org/apache/camel/component/alibaba/oss/OSSConsumerInProgressTest.java
b/components/camel-alibaba/camel-alibaba-oss/src/test/java/org/apache/camel/component/alibaba/oss/OSSConsumerInProgressTest.java
new file mode 100644
index 000000000000..8125bdc8f74a
--- /dev/null
+++
b/components/camel-alibaba/camel-alibaba-oss/src/test/java/org/apache/camel/component/alibaba/oss/OSSConsumerInProgressTest.java
@@ -0,0 +1,280 @@
+/*
+ * 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.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+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.RuntimeCamelException;
+import org.apache.camel.ServiceStatus;
+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 org.mockito.Mockito;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+/**
+ * The consumer keeps the objects whose exchanges are being processed: an
object still in flight is not consumed again
+ * by the next poll, an object is consumed again once its exchange failed or
completed (without deleteAfterRead), an
+ * object deleted after the listing is skipped without failing the poll, and
the objects of a batch that a stopping
+ * consumer does not process are consumed when the route is started again.
+ */
+class OSSConsumerInProgressTest extends CamelTestSupport {
+
+ private final Set<String> objects = new ConcurrentSkipListSet<>();
+ private final Set<String> deletedAfterListing =
ConcurrentHashMap.newKeySet();
+ private final Map<String, AtomicInteger> gets = new ConcurrentHashMap<>();
+ private final AtomicInteger failedPolls = new AtomicInteger();
+ private final AtomicInteger attempts = new AtomicInteger();
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("ossClient", ossClient());
+ // counts the polls that completed (the exchanges of the poll were
handed to the route) and the failed polls
+ 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) {
+ failedPolls.incrementAndGet();
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void releaseRoute() {
+ release.countDown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:hold")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+
+ from("seda:failOnce")
+ .process(e -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("first attempt
fails");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void objectInFlightIsNotConsumedAgain() throws Exception {
+ objects.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a.txt");
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls",
"seda:hold");
+
+ assertThat(inRoute.await(20, TimeUnit.SECONDS)).as("the object was not
consumed").isTrue();
+ // the second poll has completed while the first exchange is still
held in the seda route
+ assertThat(twoPolls.await(20, TimeUnit.SECONDS)).as("the consumer did
not poll again").isTrue();
+ int getsWhileInFlight = gets("a.txt");
+ release.countDown();
+
+ assertThat(getsWhileInFlight)
+ .as("a.txt was consumed again by a later poll while its first
exchange was still in flight")
+ .isEqualTo(1);
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void failedObjectIsConsumedAgain() throws Exception {
+ objects.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a.txt");
+
+ addConsumerRoute("&deleteAfterRead=true", "seda:failOnce");
+
+ mock.assertIsSatisfied();
+ assertThat(attempts).as("a.txt should be consumed again after its
first exchange failed").hasValue(2);
+ }
+
+ @Test
+ void completedObjectIsConsumedAgainWithoutDelete() throws Exception {
+ objects.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(2);
+ mock.allMessages().body().isEqualTo("a.txt");
+
+ addConsumerRoute("", "mock:result");
+
+ // without deleteAfterRead every poll reads the object again once its
previous exchange is done
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void deletedObjectIsSkipped() throws Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ objects.add("c.txt");
+ deletedAfterListing.add("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a.txt", "c.txt");
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls",
"mock:result");
+
+ mock.assertIsSatisfied();
+ assertThat(failedPolls).as("the poll failed because b.txt was deleted
after the listing").hasValue(0);
+ assertThat(gets("a.txt")).as("a.txt was read again because the poll
that listed b.txt failed").isEqualTo(1);
+ }
+
+ @Test
+ void objectsOfAStoppedBatchAreConsumedAfterRestart() throws Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ objects.add("c.txt");
+ AtomicBoolean first = new AtomicBoolean(true);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("alibaba-oss:bucket?ossClient=#ossClient&initialDelay=0&delay=10&deleteAfterRead=true").routeId("oss")
+ .process(e -> {
+ if (first.compareAndSet(true, false)) {
+ // stop the route while the first exchange of
the batch is processed, and wait until
+ // the consumer does not process the rest of
the batch
+ OSSConsumer consumer = (OSSConsumer)
context.getRoute("oss").getConsumer();
+ CompletableFuture.runAsync(() -> {
+ try {
+
context.getRouteController().stopRoute("oss");
+ } catch (Exception ex) {
+ throw new RuntimeCamelException(ex);
+ }
+ });
+ await().atMost(20, TimeUnit.SECONDS).until(()
-> !consumer.isBatchAllowed());
+ }
+ })
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ mock.expectedBodiesReceived("a.txt");
+ mock.assertIsSatisfied();
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() ->
context.getRouteController().getRouteStatus("oss") == ServiceStatus.Stopped);
+
+ mock.reset();
+ mock.expectedBodiesReceivedInAnyOrder("b.txt", "c.txt");
+ context.getRouteController().startRoute("oss");
+
+ mock.assertIsSatisfied();
+ }
+
+ private void addConsumerRoute(String options, String to) 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(to);
+ }
+ });
+ }
+
+ private int gets(String key) {
+ AtomicInteger n = gets.get(key);
+ return n != null ? n.get() : 0;
+ }
+
+ private OSSClient ossClient() {
+ OSSClient client = Mockito.mock(OSSClient.class);
+
Mockito.when(client.listObjectsV2(Mockito.any(ListObjectsV2Request.class))).thenAnswer(i
-> {
+ List<ObjectSummary> contents = objects.stream().map(key -> {
+ ObjectSummary summary = Mockito.mock(ObjectSummary.class);
+ Mockito.when(summary.key()).thenReturn(key);
+ return summary;
+ }).toList();
+ // deleted (as if by another consumer) right after the listing
returned it
+ objects.removeAll(deletedAfterListing);
+ ListObjectsV2Result result =
Mockito.mock(ListObjectsV2Result.class);
+ Mockito.when(result.contents()).thenReturn(contents);
+ 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());
+ }
+ gets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
+ 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/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
b/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
index 3a7811d62dc0..134aa102e3c3 100644
---
a/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
+++
b/components/camel-aws/camel-aws2-s3-vectors/src/main/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumer.java
@@ -17,11 +17,11 @@
package org.apache.camel.component.aws2.s3vectors;
import java.util.ArrayList;
-import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import org.apache.camel.AsyncCallback;
import org.apache.camel.Exchange;
@@ -47,7 +47,10 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(AWS2S3VectorsConsumer.class);
- private final Set<String> processedVectorIds = new HashSet<>();
+ // vectors already consumed (without deleteAfterRead) or whose exchanges
are being processed (with
+ // deleteAfterRead), which a later poll must not consume again. It is
updated from the threads that complete the
+ // exchanges, so it must be thread-safe.
+ private final Set<String> processedVectorIds =
ConcurrentHashMap.newKeySet();
public AWS2S3VectorsConsumer(AWS2S3VectorsEndpoint endpoint, Processor
processor) {
super(endpoint, processor);
@@ -103,8 +106,8 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
for (QueryOutputVector vector : vectors) {
String vectorId = vector.key();
- // Skip if already processed (to avoid duplicates in
subsequent polls)
- if (processedVectorIds.contains(vectorId)) {
+ // Skip if already processed or still in flight (to avoid
duplicates in subsequent polls)
+ if (vectorId != null && !processedVectorIds.add(vectorId))
{
LOG.trace("Skipping already processed vector [{}]",
vectorId);
continue;
}
@@ -123,15 +126,16 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
// Add delete callback if deleteAfterRead is enabled
if (getConfiguration().isDeleteAfterRead()) {
+ // the vector is tracked only while its exchange is in
flight, as a deleted vector cannot be
+ // returned again, and one that failed must be retried
on a subsequent poll
exchange.getExchangeExtension().addOnCompletion(
new VectorDeleteSynchronization(
getS3VectorsClient(),
vectorBucketName, vectorIndexName,
vectorId));
} else {
- // Track for de-duplication only when we are not
deleting (a deleted vector cannot be
- // returned again). Mark it now to avoid re-delivering
it on overlapping polls, but drop it
- // again if the exchange fails so it can be retried on
a subsequent poll.
- processedVectorIds.add(vectorId);
+ // Track for de-duplication when we are not deleting,
to avoid re-delivering the vector on
+ // later polls, but drop it again if the exchange
fails so it can be retried on a subsequent
+ // poll.
exchange.getExchangeExtension().addOnCompletion(new
VectorDedupSynchronization(vectorId));
}
@@ -168,6 +172,13 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+ // the remaining exchanges are not processed as the consumer is
stopping, so a later poll must consume their
+ // vectors again
+ for (Object exchange : exchanges) {
+ releaseVector(((Exchange)
exchange).getIn().getHeader(AWS2S3VectorsConstants.VECTOR_ID, String.class));
+ releaseExchange((Exchange) exchange, false);
+ }
+ exchanges.clear();
return total;
}
@@ -238,10 +249,17 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
return result;
}
+ private void releaseVector(String vectorId) {
+ if (vectorId != null) {
+ processedVectorIds.remove(vectorId);
+ }
+ }
+
/**
- * Synchronization callback to delete a vector after successful processing
+ * Synchronization callback to delete a vector after successful
processing. The vector is no longer in flight once
+ * its exchange is done, whether it was deleted or failed.
*/
- private static class VectorDeleteSynchronization extends
SynchronizationAdapter {
+ private class VectorDeleteSynchronization extends SynchronizationAdapter {
private final S3VectorsClient client;
private final String vectorBucketName;
@@ -275,12 +293,15 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
LOG.debug("Deleted vector [{}] after successful processing",
vectorId);
} catch (Exception e) {
LOG.warn("Error deleting vector [{}] after processing: {}",
vectorId, e.getMessage(), e);
+ } finally {
+ releaseVector(vectorId);
}
}
@Override
public void onFailure(Exchange exchange) {
LOG.trace("Exchange failed, not deleting vector [{}]", vectorId);
+ releaseVector(vectorId);
}
}
@@ -298,7 +319,7 @@ public class AWS2S3VectorsConsumer extends
ScheduledBatchPollingConsumer {
@Override
public void onFailure(Exchange exchange) {
- processedVectorIds.remove(vectorId);
+ releaseVector(vectorId);
}
}
}
diff --git
a/components/camel-aws/camel-aws2-s3-vectors/src/test/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumerInProgressTest.java
b/components/camel-aws/camel-aws2-s3-vectors/src/test/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumerInProgressTest.java
new file mode 100644
index 000000000000..b89d50f3cd93
--- /dev/null
+++
b/components/camel-aws/camel-aws2-s3-vectors/src/test/java/org/apache/camel/component/aws2/s3vectors/AWS2S3VectorsConsumerInProgressTest.java
@@ -0,0 +1,163 @@
+/*
+ * 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.aws2.s3vectors;
+
+import java.util.Set;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+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 org.mockito.Mockito;
+import software.amazon.awssdk.services.s3vectors.S3VectorsClient;
+import software.amazon.awssdk.services.s3vectors.model.DeleteVectorsRequest;
+import software.amazon.awssdk.services.s3vectors.model.DeleteVectorsResponse;
+import software.amazon.awssdk.services.s3vectors.model.QueryOutputVector;
+import software.amazon.awssdk.services.s3vectors.model.QueryVectorsRequest;
+import software.amazon.awssdk.services.s3vectors.model.QueryVectorsResponse;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * With deleteAfterRead the consumer keeps the vectors whose exchanges are in
flight: such a vector is not consumed
+ * again by the next poll, and a vector whose exchange failed is consumed
again.
+ */
+class AWS2S3VectorsConsumerInProgressTest extends CamelTestSupport {
+
+ private final Set<String> vectors = new ConcurrentSkipListSet<>();
+ private final AtomicInteger consumed = new AtomicInteger();
+ private final AtomicInteger attempts = new AtomicInteger();
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ S3VectorsClient client = Mockito.mock(S3VectorsClient.class);
+
Mockito.when(client.queryVectors(Mockito.any(QueryVectorsRequest.class))).thenAnswer(i
-> QueryVectorsResponse
+ .builder()
+ .vectors(vectors.stream().map(key ->
QueryOutputVector.builder().key(key).build()).toList())
+ .build());
+
Mockito.when(client.deleteVectors(Mockito.any(DeleteVectorsRequest.class))).thenAnswer(i
-> {
+ i.<DeleteVectorsRequest>
getArgument(0).keys().forEach(vectors::remove);
+ return DeleteVectorsResponse.builder().build();
+ });
+ context.getRegistry().bind("client", client);
+ // 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 releaseRoute() {
+ release.countDown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:hold")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+
+ from("seda:failOnce")
+ .process(e -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("first attempt
fails");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void vectorInFlightIsNotConsumedAgain() throws Exception {
+ vectors.add("v1");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedHeaderReceived(AWS2S3VectorsConstants.VECTOR_ID, "v1");
+
+ addConsumerRoute("&pollStrategy=#countPolls", "seda:hold");
+
+ assertThat(inRoute.await(20, TimeUnit.SECONDS)).as("the vector was not
consumed").isTrue();
+ // the second poll has completed while the first exchange is still
held in the seda route
+ assertThat(twoPolls.await(20, TimeUnit.SECONDS)).as("the consumer did
not poll again").isTrue();
+ int consumedWhileInFlight = consumed.get();
+ release.countDown();
+
+ assertThat(consumedWhileInFlight)
+ .as("v1 was consumed again by a later poll while its first
exchange was still in flight")
+ .isEqualTo(1);
+ mock.assertIsSatisfied();
+ assertThat(vectors).as("v1 should be deleted after its exchange
completed").isEmpty();
+ }
+
+ @Test
+ void failedVectorIsConsumedAgain() throws Exception {
+ vectors.add("v1");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedHeaderReceived(AWS2S3VectorsConstants.VECTOR_ID, "v1");
+
+ addConsumerRoute("", "seda:failOnce");
+
+ mock.assertIsSatisfied();
+ assertThat(attempts).as("v1 should be consumed again after its first
exchange failed").hasValue(2);
+ }
+
+ private void addConsumerRoute(String options, String to) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("aws2-s3-vectors://bucket?vectorIndexName=index&s3VectorsClient=#client&consumerQueryVector=0.1,0.2"
+ + "&deleteAfterRead=true&initialDelay=0&delay=10" +
options)
+ .process(e -> consumed.incrementAndGet())
+ .to(to);
+ }
+ });
+ }
+}
diff --git
a/components/camel-azure/camel-azure-storage-blob/src/main/docs/azure-storage-blob-consumer.adoc
b/components/camel-azure/camel-azure-storage-blob/src/main/docs/azure-storage-blob-consumer.adoc
index 278efa914273..a15d2b697f50 100644
---
a/components/camel-azure/camel-azure-storage-blob/src/main/docs/azure-storage-blob-consumer.adoc
+++
b/components/camel-azure/camel-azure-storage-blob/src/main/docs/azure-storage-blob-consumer.adoc
@@ -171,6 +171,8 @@ IMPORTANT: The delete operation is only performed if the
Exchange is successfull
NOTE: When `deleteAfterRead` is set to `false` (the default), the same blobs
will be retrieved repeatedly in subsequent polls. In this case, you should use
the Idempotent Consumer EIP to filter out duplicates based on the
`CamelAzureStorageBlobBlobName` header.
+NOTE: A blob is not retrieved again while the exchange of a previous poll is
still processing it, for example when the route continues asynchronously (such
as `to("seda:...")`). It is retrieved again by a later poll once that exchange
is done, unless it was deleted or moved. A blob that was deleted (for example
by another consumer) between the listing and its download is skipped.
+
=== Move After Read
The consumer can move blobs to a different container after successful
processing. This is useful for archiving processed blobs or implementing a
processing pipeline across containers.
diff --git
a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java
b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java
index 89a16bf70561..c08ff5a44bea 100644
---
a/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java
+++
b/components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/BlobConsumer.java
@@ -16,11 +16,14 @@
*/
package org.apache.camel.component.azure.storage.blob;
+import java.io.Closeable;
import java.io.IOException;
import java.time.Duration;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import com.azure.storage.blob.BlobClient;
import com.azure.storage.blob.BlobContainerClient;
@@ -38,6 +41,7 @@ import
org.apache.camel.component.azure.storage.blob.operations.BlobOperations;
import org.apache.camel.spi.Synchronization;
import org.apache.camel.support.ScheduledBatchPollingConsumer;
import org.apache.camel.util.CastUtils;
+import org.apache.camel.util.IOHelper;
import org.apache.camel.util.ObjectHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,6 +50,9 @@ public class BlobConsumer extends
ScheduledBatchPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(BlobConsumer.class);
+ // blobs whose exchanges are being processed, which a later poll must not
consume again
+ private final Set<String> inProgress = ConcurrentHashMap.newKeySet();
+
public BlobConsumer(final BlobEndpoint endpoint, final Processor
processor) {
super(endpoint, processor);
}
@@ -75,9 +82,8 @@ public class BlobConsumer extends
ScheduledBatchPollingConsumer {
try {
if (ObjectHelper.isNotEmpty(blobName)) {
// here we have a blob which means we just download a single
blob
- final Exchange exchange = createExchangeFromBlob(blobName,
blobContainerClient);
exchanges = new LinkedList<>();
- exchanges.add(exchange);
+ addExchange(blobName, blobContainerClient, exchanges, false);
} else {
// download multiple blobs since we only have no blobName set
exchanges =
createBatchExchangesFromContainer(blobContainerClient);
@@ -127,12 +133,59 @@ public class BlobConsumer extends
ScheduledBatchPollingConsumer {
forceConsumerAsReady();
final Queue<Exchange> exchanges = new LinkedList<>();
- for (BlobItem blobItem : blobs) {
- exchanges.add(createExchangeFromBlob(blobItem.getName(),
blobContainerClient));
+ try {
+ for (BlobItem blobItem : blobs) {
+ addExchange(blobItem.getName(), blobContainerClient,
exchanges, true);
+ }
+ } catch (Exception e) {
+ // none of the blobs of this poll is processed
+ releaseNotProcessed(exchanges);
+ throw e;
}
return exchanges;
}
+ private void addExchange(
+ final String blobName, final BlobContainerClient
blobContainerClient, final Queue<Exchange> exchanges,
+ final boolean listed)
+ throws IOException {
+ // skip the blob if an exchange of a previous poll is still processing
it
+ if (!inProgress.add(blobName)) {
+ LOG.trace("Skipping blob {} as it is already in progress",
blobName);
+ return;
+ }
+ try {
+ exchanges.add(createExchangeFromBlob(blobName,
blobContainerClient));
+ } catch (Exception e) {
+ inProgress.remove(blobName);
+ if (listed && e instanceof BlobStorageException bse &&
bse.getStatusCode() == 404) {
+ // deleted (for example by another consumer) after it was
listed
+ LOG.debug("Skipping blob {} as it no longer exists", blobName);
+ return;
+ }
+ throw e;
+ }
+ }
+
+ /**
+ * Releases the exchanges that are not routed: their blobs are no longer
in progress, so a later poll consumes them
+ * again, and their blob input streams are closed, as no on completion
does it.
+ */
+ private void releaseNotProcessed(final Queue<?> exchanges) {
+ for (Object o : exchanges) {
+ final Exchange exchange = (Exchange) o;
+ final String name =
exchange.getIn().getHeader(BlobConstants.BLOB_NAME, String.class);
+ if (name != null) {
+ inProgress.remove(name);
+ }
+ if (exchange.getIn().getBody() instanceof Closeable body) {
+ IOHelper.close(body);
+ }
+ releaseExchange(exchange, true);
+ }
+ exchanges.clear();
+ }
+
@Override
public BlobEndpoint getEndpoint() {
return (BlobEndpoint) super.getEndpoint();
@@ -154,16 +207,25 @@ public class BlobConsumer extends
ScheduledBatchPollingConsumer {
// update pending number of exchanges
pendingExchanges = total - index - 1;
+ final String blobName =
exchange.getIn().getHeader(BlobConstants.BLOB_NAME, String.class);
// add on completion to handle after work when the exchange is done
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
@Override
public void onComplete(Exchange exchange) {
- processCommit(exchange);
+ try {
+ processCommit(exchange);
+ } finally {
+ releaseInProgress(blobName);
+ }
}
@Override
public void onFailure(Exchange exchange) {
- processRollback(exchange);
+ try {
+ processRollback(exchange);
+ } finally {
+ releaseInProgress(blobName);
+ }
}
});
@@ -171,9 +233,17 @@ public class BlobConsumer extends
ScheduledBatchPollingConsumer {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
+ // the remaining exchanges are not processed as the consumer is
stopping
+ releaseNotProcessed(exchanges);
return total;
}
+ private void releaseInProgress(final String blobName) {
+ if (blobName != null) {
+ inProgress.remove(blobName);
+ }
+ }
+
/**
* Strategy when processing the exchange failed.
*
diff --git
a/components/camel-azure/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerInProgressTest.java
b/components/camel-azure/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerInProgressTest.java
new file mode 100644
index 000000000000..26d8c4d4f87f
--- /dev/null
+++
b/components/camel-azure/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerInProgressTest.java
@@ -0,0 +1,277 @@
+/*
+ * 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.azure.storage.blob;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import com.azure.core.http.HttpResponse;
+import com.azure.core.http.rest.PagedIterable;
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClient;
+import com.azure.storage.blob.models.BlobItem;
+import com.azure.storage.blob.models.BlobProperties;
+import com.azure.storage.blob.models.BlobRange;
+import com.azure.storage.blob.models.BlobRequestConditions;
+import com.azure.storage.blob.models.BlobStorageException;
+import com.azure.storage.blob.specialized.BlobInputStream;
+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;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The consumer keeps the blobs whose exchanges are being processed: a blob
still in flight is not consumed again by the
+ * next poll, a blob is consumed again once its exchange failed or completed
(without deleteAfterRead), a blob deleted
+ * after the listing is skipped without failing the poll, and the blobs of a
batch that a stopping consumer does not
+ * process are released and their streams closed.
+ */
+class BlobConsumerInProgressTest extends CamelTestSupport {
+
+ private final Set<String> blobs = new ConcurrentSkipListSet<>();
+ private final Set<String> deletedAfterListing =
ConcurrentHashMap.newKeySet();
+ private final Map<String, AtomicInteger> reads = new ConcurrentHashMap<>();
+ private final List<BlobInputStream> streams = new CopyOnWriteArrayList<>();
+ private final AtomicInteger failedPolls = new AtomicInteger();
+ private final AtomicInteger attempts = new AtomicInteger();
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("client", serviceClient());
+ // counts the polls that completed (the exchanges of the poll were
handed to the route) and the failed polls
+ 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) {
+ failedPolls.incrementAndGet();
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void releaseRoute() {
+ release.countDown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:hold")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+
+ from("seda:failOnce")
+ .process(e -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("first attempt
fails");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ void blobInFlightIsNotConsumedAgain() throws Exception {
+ assertNotConsumedAgain("");
+ }
+
+ @Test
+ void blobInFlightIsNotConsumedAgainWithBlobName() throws Exception {
+ assertNotConsumedAgain("&blobName=a.txt");
+ }
+
+ @Test
+ void failedBlobIsConsumedAgain() throws Exception {
+ blobs.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedHeaderReceived(BlobConstants.BLOB_NAME, "a.txt");
+
+ addConsumerRoute("&deleteAfterRead=true", "seda:failOnce");
+
+ mock.assertIsSatisfied();
+ assertEquals(2, attempts.get(), "a.txt should be consumed again after
its first exchange failed");
+ }
+
+ @Test
+ void completedBlobIsConsumedAgainWithoutDelete() throws Exception {
+ blobs.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(2);
+ mock.allMessages().header(BlobConstants.BLOB_NAME).isEqualTo("a.txt");
+
+ addConsumerRoute("", "mock:result");
+
+ // without deleteAfterRead every poll reads the blob again once its
previous exchange is done
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ void deletedBlobIsSkipped() throws Exception {
+ blobs.add("a.txt");
+ blobs.add("b.txt");
+ blobs.add("c.txt");
+ deletedAfterListing.add("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedHeaderValuesReceivedInAnyOrder(BlobConstants.BLOB_NAME,
"a.txt", "c.txt");
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls",
"mock:result");
+
+ mock.assertIsSatisfied();
+ // the poll that listed b.txt is not dropped, so the blobs read before
b.txt are not read again by a later poll
+ assertEquals(1, reads("a.txt"), "a.txt was read again because b.txt
was deleted after the listing");
+ assertEquals(1, reads("c.txt"));
+ assertEquals(0, failedPolls.get(), "the poll failed because b.txt was
deleted after the listing");
+ }
+
+ @Test
+ void notProcessedBlobIsReleased() throws Exception {
+ blobs.add("a.txt");
+ BlobEndpoint endpoint = context.getEndpoint(
+
"azure-storage-blob://account/container?serviceClient=#client&deleteAfterRead=true",
BlobEndpoint.class);
+ // never started, so the consumer does not process any exchange of the
batch
+ BlobConsumer consumer = (BlobConsumer)
endpoint.createConsumer(exchange -> {
+ throw new IllegalStateException("no exchange should be processed");
+ });
+
+ consumer.poll();
+ consumer.poll();
+
+ assertEquals(2, reads("a.txt"), "a.txt stayed in progress although its
exchange was not processed");
+ assertEquals(2, streams.size());
+ for (BlobInputStream stream : streams) {
+ verify(stream).close();
+ }
+ }
+
+ private void assertNotConsumedAgain(String options) throws Exception {
+ blobs.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedHeaderReceived(BlobConstants.BLOB_NAME, "a.txt");
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls" +
options, "seda:hold");
+
+ assertTrue(inRoute.await(20, TimeUnit.SECONDS), "the blob 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 readsWhileInFlight = reads("a.txt");
+ release.countDown();
+
+ assertEquals(1, readsWhileInFlight,
+ "a.txt was consumed again by a later poll while its first
exchange was still in flight");
+ mock.assertIsSatisfied();
+ }
+
+ private void addConsumerRoute(String options, String to) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("azure-storage-blob://account/container?serviceClient=#client&initialDelay=0&delay=10"
+ options)
+ .to(to);
+ }
+ });
+ }
+
+ private int reads(String name) {
+ AtomicInteger n = reads.get(name);
+ return n != null ? n.get() : 0;
+ }
+
+ @SuppressWarnings("unchecked")
+ private BlobServiceClient serviceClient() {
+ BlobServiceClient serviceClient = mock(BlobServiceClient.class);
+ BlobContainerClient containerClient = mock(BlobContainerClient.class);
+
when(serviceClient.getBlobContainerClient(anyString())).thenReturn(containerClient);
+
+ PagedIterable<BlobItem> listing = mock(PagedIterable.class);
+ when(listing.stream()).thenAnswer(i -> {
+ List<BlobItem> items = blobs.stream().map(name -> new
BlobItem().setName(name)).toList();
+ // deleted (as if by another consumer) right after the listing
returned it
+ blobs.removeAll(deletedAfterListing);
+ return items.stream();
+ });
+ when(containerClient.listBlobs(any(), any())).thenReturn(listing);
+
+ when(containerClient.getBlobClient(anyString())).thenAnswer(i ->
blobClient(i.getArgument(0)));
+ return serviceClient;
+ }
+
+ private BlobClient blobClient(String name) {
+ BlobClient blobClient = mock(BlobClient.class);
+ when(blobClient.openInputStream(nullable(BlobRange.class),
nullable(BlobRequestConditions.class))).thenAnswer(i -> {
+ if (!blobs.contains(name)) {
+ HttpResponse response = mock(HttpResponse.class);
+ when(response.getStatusCode()).thenReturn(404);
+ throw new BlobStorageException("The specified blob does not
exist.", response, null);
+ }
+ reads.computeIfAbsent(name, k -> new
AtomicInteger()).incrementAndGet();
+ BlobInputStream stream = mock(BlobInputStream.class);
+
when(stream.getProperties()).thenReturn(mock(BlobProperties.class));
+ streams.add(stream);
+ return stream;
+ });
+ doAnswer(i -> {
+ blobs.remove(name);
+ return null;
+ }).when(blobClient).delete();
+ return blobClient;
+ }
+}
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 79d2b361b34d..385fb1e4f479 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,10 +16,11 @@
*/
package org.apache.camel.component.huaweicloud.obs;
-import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import com.obs.services.ObsClient;
import com.obs.services.exception.ObsException;
@@ -44,6 +45,8 @@ public class OBSConsumer extends
ScheduledBatchPollingConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(OBSConsumer.class.getName());
private final OBSEndpoint endpoint;
+ // objects whose exchanges are being processed, which a later poll must
not consume again
+ private final Set<String> inProgress = ConcurrentHashMap.newKeySet();
private ObsClient obsClient;
private String marker;
private boolean destinationBucketCreated;
@@ -117,10 +120,20 @@ public class OBSConsumer extends
ScheduledBatchPollingConsumer {
exchanges = new LinkedList<>();
} else if (ObjectHelper.isNotEmpty(fileName)) {
// when file name is given, just download that object
- ObsObject object = obsClient.getObject(bucketName, fileName);
- List<ObsObject> list = new ArrayList<>();
- list.add(object);
- exchanges = createExchanges(list);
+ exchanges = new LinkedList<>();
+ // skip the object if an exchange of a previous poll is still
processing it
+ if (inProgress.add(fileName)) {
+ ObsObject object;
+ try {
+ object = obsClient.getObject(bucketName, fileName);
+ } catch (RuntimeException e) {
+ inProgress.remove(fileName);
+ throw e;
+ }
+ addExchange(fileName, object, exchanges);
+ } else {
+ LOG.trace("Skipping object {} as it is already in progress",
fileName);
+ }
} else {
// if file name is not given, get a list of all objects
ListObjectsRequest request = new ListObjectsRequest(bucketName);
@@ -169,21 +182,32 @@ public class OBSConsumer extends
ScheduledBatchPollingConsumer {
// update number of pending exchanges
pendingExchanges = total - index - 1;
+ final String objectKey =
exchange.getIn().getHeader(OBSHeaders.OBJECT_KEY, String.class);
exchange.getExchangeExtension().addOnCompletion(new
Synchronization() {
@Override
public void onComplete(Exchange exchange) {
- processComplete(exchange);
+ try {
+ processComplete(exchange);
+ } finally {
+ inProgress.remove(objectKey);
+ }
}
@Override
public void onFailure(Exchange exchange) {
- processFailure(exchange);
+ try {
+ processFailure(exchange);
+ } finally {
+ inProgress.remove(objectKey);
+ }
}
});
AsyncCallback callback = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, callback);
}
+ // the remaining exchanges are not processed as the consumer is
stopping
+ releaseNotProcessed(exchanges);
return total;
}
@@ -193,12 +217,44 @@ public class OBSConsumer extends
ScheduledBatchPollingConsumer {
*/
private Queue<Exchange> createExchanges(List<ObsObject> obsObjects) {
Queue<Exchange> answer = new LinkedList<>();
- for (ObsObject objectSummary : obsObjects) {
+ 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);
+ 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;
+ }
+ }
+ } catch (RuntimeException e) {
+ // none of the objects of this poll is processed
+ releaseNotProcessed(answer);
+ throw e;
+ }
+ return answer;
+ }
+
+ /**
+ * Adds an exchange for the object, which is in progress, to the answer.
The object is no longer in progress if no
+ * exchange is added.
+ */
+ private void addExchange(String objectKey, ObsObject objectSummary,
Queue<Exchange> answer) {
+ try {
ObsObject obsObject;
if (objectSummary.getMetadata().getContentType() == null) {
// object was from list objects. Since not all object data is
included when listing objects, we must retrieve all the data by calling
getObject
- obsObject = obsClient.getObject(endpoint.getBucketName(),
objectSummary.getObjectKey());
+ obsObject = obsClient.getObject(endpoint.getBucketName(),
objectKey);
} else {
// object was already retrieved using getObjects
obsObject = objectSummary;
@@ -208,9 +264,25 @@ public class OBSConsumer extends
ScheduledBatchPollingConsumer {
if (includeObsObject(obsObject)) {
Exchange exchange = createExchange(obsObject);
answer.add(exchange);
+ } else {
+ inProgress.remove(objectKey);
}
+ } catch (RuntimeException e) {
+ inProgress.remove(objectKey);
+ throw e;
}
- return answer;
+ }
+
+ private void releaseNotProcessed(Queue<?> exchanges) {
+ for (Object o : exchanges) {
+ Exchange exchange = (Exchange) o;
+ String objectKey =
exchange.getIn().getHeader(OBSHeaders.OBJECT_KEY, String.class);
+ if (objectKey != null) {
+ inProgress.remove(objectKey);
+ }
+ releaseExchange(exchange, true);
+ }
+ exchanges.clear();
}
/**
diff --git
a/components/camel-huawei/camel-huaweicloud-obs/src/test/java/org/apache/camel/component/huaweicloud/obs/OBSConsumerInProgressTest.java
b/components/camel-huawei/camel-huaweicloud-obs/src/test/java/org/apache/camel/component/huaweicloud/obs/OBSConsumerInProgressTest.java
new file mode 100644
index 000000000000..7e09843ffb11
--- /dev/null
+++
b/components/camel-huawei/camel-huaweicloud-obs/src/test/java/org/apache/camel/component/huaweicloud/obs/OBSConsumerInProgressTest.java
@@ -0,0 +1,300 @@
+/*
+ * 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.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentSkipListSet;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+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.RuntimeCamelException;
+import org.apache.camel.ServiceStatus;
+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 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;
+
+/**
+ * The consumer keeps the objects whose exchanges are being processed: an
object still in flight is not consumed again
+ * by the next poll, an object is consumed again once its exchange failed or
completed (without deleteAfterRead), an
+ * object deleted after the listing is skipped without failing the poll, and
the objects of a batch that a stopping
+ * consumer does not process are consumed when the route is started again.
+ */
+public class OBSConsumerInProgressTest extends CamelTestSupport {
+
+ private static final String BUCKET = "bucket";
+
+ private final Set<String> objects = new ConcurrentSkipListSet<>();
+ private final Set<String> deletedAfterListing =
ConcurrentHashMap.newKeySet();
+ private final Map<String, AtomicInteger> gets = new ConcurrentHashMap<>();
+ private final AtomicInteger failedPolls = new AtomicInteger();
+ private final AtomicInteger attempts = new AtomicInteger();
+ private final CountDownLatch inRoute = new CountDownLatch(1);
+ private final CountDownLatch twoPolls = new CountDownLatch(2);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("obsClient", obsClient());
+ // counts the polls that completed (the exchanges of the poll were
handed to the route) and the failed polls
+ 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) {
+ failedPolls.incrementAndGet();
+ return false;
+ }
+ });
+ return context;
+ }
+
+ @AfterEach
+ void releaseRoute() {
+ release.countDown();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ // seda hands the on completions (the delete) over to this
route
+ from("seda:hold")
+ .process(e -> {
+ inRoute.countDown();
+ release.await(20, TimeUnit.SECONDS);
+ })
+ .to("mock:result");
+
+ from("seda:failOnce")
+ .process(e -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new IllegalStateException("first attempt
fails");
+ }
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ @Test
+ public void objectInFlightIsNotConsumedAgain() throws Exception {
+ assertNotConsumedAgain("");
+ }
+
+ @Test
+ public void objectInFlightIsNotConsumedAgainWithFileName() throws
Exception {
+ assertNotConsumedAgain("&fileName=a.txt");
+ }
+
+ @Test
+ public void failedObjectIsConsumedAgain() throws Exception {
+ objects.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a.txt");
+
+ addConsumerRoute("&deleteAfterRead=true", "seda:failOnce");
+
+ mock.assertIsSatisfied();
+ assertEquals(2, attempts.get(), "a.txt should be consumed again after
its first exchange failed");
+ }
+
+ @Test
+ public void completedObjectIsConsumedAgainWithoutDelete() throws Exception
{
+ objects.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(2);
+ mock.allMessages().body().isEqualTo("a.txt");
+
+ addConsumerRoute("", "mock:result");
+
+ // without deleteAfterRead every poll reads the object again once its
previous exchange is done
+ mock.assertIsSatisfied();
+ }
+
+ @Test
+ public void deletedObjectIsSkipped() throws Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ objects.add("c.txt");
+ deletedAfterListing.add("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceivedInAnyOrder("a.txt", "c.txt");
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls",
"mock:result");
+
+ mock.assertIsSatisfied();
+ assertEquals(0, failedPolls.get(), "the poll failed because b.txt was
deleted after the listing");
+ assertEquals(1, gets("a.txt"), "a.txt was read again because the poll
that listed b.txt failed");
+ }
+
+ @Test
+ public void objectsOfAStoppedBatchAreConsumedAfterRestart() throws
Exception {
+ objects.add("a.txt");
+ objects.add("b.txt");
+ objects.add("c.txt");
+ AtomicBoolean first = new AtomicBoolean(true);
+ MockEndpoint mock = getMockEndpoint("mock:result");
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(obsUri("&deleteAfterRead=true")).routeId("obs")
+ .process(e -> {
+ if (first.compareAndSet(true, false)) {
+ // stop the route while the first exchange of
the batch is processed, and wait until
+ // the consumer does not process the rest of
the batch
+ OBSConsumer consumer = (OBSConsumer)
context.getRoute("obs").getConsumer();
+ CompletableFuture.runAsync(() -> {
+ try {
+
context.getRouteController().stopRoute("obs");
+ } catch (Exception ex) {
+ throw new RuntimeCamelException(ex);
+ }
+ });
+ await().atMost(20, TimeUnit.SECONDS).until(()
-> !consumer.isBatchAllowed());
+ }
+ })
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ });
+
+ mock.expectedBodiesReceived("a.txt");
+ mock.assertIsSatisfied();
+ await().atMost(20, TimeUnit.SECONDS)
+ .until(() ->
context.getRouteController().getRouteStatus("obs") == ServiceStatus.Stopped);
+
+ mock.reset();
+ mock.expectedBodiesReceivedInAnyOrder("b.txt", "c.txt");
+ context.getRouteController().startRoute("obs");
+
+ mock.assertIsSatisfied();
+ }
+
+ private void assertNotConsumedAgain(String options) throws Exception {
+ objects.add("a.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("a.txt");
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls" +
options, "seda:hold");
+
+ 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 getsWhileInFlight = gets("a.txt");
+ release.countDown();
+
+ assertEquals(1, getsWhileInFlight,
+ "a.txt was consumed again by a later poll while its first
exchange was still in flight");
+ mock.assertIsSatisfied();
+ }
+
+ private String obsUri(String options) {
+ return "hwcloud-obs:?bucketName=" + BUCKET +
"&obsClient=#obsClient&initialDelay=0&delay=10" + options;
+ }
+
+ private void addConsumerRoute(String options, String to) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(obsUri(options))
+ .convertBodyTo(String.class)
+ .to(to);
+ }
+ });
+ }
+
+ private int gets(String key) {
+ AtomicInteger n = gets.get(key);
+ return n != null ? n.get() : 0;
+ }
+
+ private ObsClient obsClient() {
+ ObsClient client = Mockito.mock(ObsClient.class);
+
Mockito.when(client.listObjects(Mockito.any(ListObjectsRequest.class))).thenAnswer(i
-> {
+ // listed objects carry no content type, so the consumer gets each
of them
+ List<ObsObject> listed = objects.stream().map(key -> {
+ ObsObject summary = new ObsObject();
+ summary.setBucketName(BUCKET);
+ summary.setObjectKey(key);
+ return summary;
+ }).toList();
+ // deleted (as if by another consumer) right after the listing
returned it
+ objects.removeAll(deletedAfterListing);
+ ObjectListing listing = Mockito.mock(ObjectListing.class);
+ Mockito.when(listing.getObjects()).thenReturn(listed);
+ 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;
+ }
+ gets.computeIfAbsent(key, k -> new
AtomicInteger()).incrementAndGet();
+ 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_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index 7c6346d6a8e7..e79e7ed43dfe 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -3662,6 +3662,26 @@ Local downloads configured with `fileDir` now resolve
existing filesystem path s
that the destination remains inside the configured directory. Downloads
through a symbolic link that
resolves outside `fileDir` are rejected. Valid nested download paths continue
to work.
+=== camel-azure-storage-blob - blobs whose exchange is in flight
+
+The consumer no longer consumes a blob again while an exchange of an earlier
poll is still processing it, which
+happened with an asynchronous route (for example `to("seda:...")`), so the
next polls now skip such a blob, and no longer
+download it again. With `deleteAfterRead=false` and `moveAfterRead=false` the
blob is consumed again by a later poll once
+its exchange is done, as before. A listed blob that no longer exists when it
is downloaded (for example deleted by
+another consumer after the listing) is now skipped, and the other blobs of the
poll are consumed, instead of the whole
+poll being dropped.
+
+=== camel-huaweicloud-obs, camel-alibaba-oss and camel-aws2-s3-vectors -
objects whose exchange is in flight
+
+The `camel-huaweicloud-obs` and `camel-alibaba-oss` consumers no longer
consume an object again while an exchange of an
+earlier poll is still processing it, which happened with an asynchronous route
(for example `to("seda:...")`), so the
+next polls now skip such an object. With `deleteAfterRead=false` (and
`moveAfterRead=false`) the object is consumed again
+by a later poll once its exchange is done, as before. A listed object that no
longer exists when it is downloaded (for
+example deleted by another consumer after the listing) is now skipped instead
of failing the poll.
+
+The `camel-aws2-s3-vectors` consumer with `deleteAfterRead=true` no longer
consumes a vector again while its exchange is
+still in flight. A vector whose exchange failed is still consumed again by a
later poll.
+
=== camel-google-sheets - custom column names of a range that does not start
at column A
The `google-sheets:application-x-struct` data type now maps the custom
`columnNames` to the columns of the range by
