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 a9ef7248b7ce CAMEL-25770: camel-azure-storage-blob - skip a listed
blob that cannot be got instead of failing the poll (#28130)
a9ef7248b7ce is described below
commit a9ef7248b7cef71e069b947892196a269bd095e8
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 20 15:16:50 2026 +0530
CAMEL-25770: camel-azure-storage-blob - skip a listed blob that cannot be
got instead of failing the poll (#28130)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/azure/storage/blob/BlobConsumer.java | 63 +++++--
.../blob/BlobConsumerBlobThatCannotBeGotTest.java | 193 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 9 +
3 files changed, 255 insertions(+), 10 deletions(-)
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 5c0158fd06c6..26e802fa75b5 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
@@ -19,8 +19,10 @@ package org.apache.camel.component.azure.storage.blob;
import java.io.Closeable;
import java.io.IOException;
import java.time.Duration;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -136,35 +138,76 @@ public class BlobConsumer extends
ScheduledBatchPollingConsumer {
forceConsumerAsReady();
final Queue<Exchange> exchanges = new LinkedList<>();
- try {
- for (BlobItem blobItem : blobs) {
- addExchange(blobItem.getName(), blobContainerClient,
exchanges, true);
+ // the listed blobs that could not be got, other than those that no
longer exist, by name
+ final Map<String, Exception> failures = new LinkedHashMap<>();
+ // whether at least one listed blob could be got
+ boolean anyGot = false;
+ for (BlobItem blobItem : blobs) {
+ // a failure to get one listed blob (access denied, a blob in the
archive tier) does not fail the poll as
+ // long as another listed blob could be got: the blob is skipped,
the other listed blobs are consumed, and
+ // a later poll tries it again
+ try {
+ anyGot |= addExchange(blobItem.getName(), blobContainerClient,
exchanges, true);
+ } catch (Exception e) {
+ failures.put(blobItem.getName(), e);
+ }
+ }
+
+ if (!failures.isEmpty()) {
+ if (!anyGot) {
+ // none of the listed blobs could be got, so this is not about
one blob (expired credentials, no
+ // permission to read blobs, no connection): the poll fails,
and the backoff of the scheduled consumer
+ // applies
+ releaseNotProcessed(exchanges);
+ final Map.Entry<String, Exception> first =
failures.entrySet().iterator().next();
+ LOG.warn("None of the listed blobs of container {} could be
got ({} failed, the first is {}): the poll fails",
+ getEndpoint().getConfiguration().getContainerName(),
failures.size(), first.getKey());
+ rethrow(first.getValue());
+ }
+ for (Map.Entry<String, Exception> failure : failures.entrySet()) {
+ getExceptionHandler().handleException("Error getting blob " +
failure.getKey() + " from container "
+ +
getEndpoint().getConfiguration().getContainerName()
+ + ". The blob is skipped
for this poll.",
+ failure.getValue());
}
- } catch (Exception e) {
- // none of the blobs of this poll is processed
- releaseNotProcessed(exchanges);
- throw e;
}
return exchanges;
}
- private void addExchange(
+ private static void rethrow(final Exception e) throws IOException {
+ if (e instanceof IOException ioe) {
+ throw ioe;
+ }
+ if (e instanceof RuntimeException re) {
+ throw re;
+ }
+ throw new IOException(e);
+ }
+
+ /**
+ * Gets a blob and adds its exchange.
+ *
+ * @return true if the blob was got and its exchange added, false if it is
skipped (an exchange of a previous poll
+ * is still processing it, or it no longer exists)
+ */
+ private boolean addExchange(
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;
+ return false;
}
try {
exchanges.add(createExchangeFromBlob(blobName,
blobContainerClient));
+ return true;
} 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;
+ return false;
}
throw e;
}
diff --git
a/components/camel-azure/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerBlobThatCannotBeGotTest.java
b/components/camel-azure/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerBlobThatCannotBeGotTest.java
new file mode 100644
index 000000000000..8d99b3dc439f
--- /dev/null
+++
b/components/camel-azure/camel-azure-storage-blob/src/test/java/org/apache/camel/component/azure/storage/blob/BlobConsumerBlobThatCannotBeGotTest.java
@@ -0,0 +1,193 @@
+/*
+ * 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.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 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.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.spi.ExceptionHandler;
+import org.apache.camel.spi.PollingConsumerPollStrategy;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import 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.when;
+
+/**
+ * A listed blob that cannot be got (a blob in the archive tier, access
denied) must not stop the consumer: the other
+ * listed blobs are consumed, and the poll fails only when none of the listed
blobs could be got. The consumer lists the
+ * whole container at every poll, so before this one such blob stopped the
consumer from consuming any blob.
+ */
+class BlobConsumerBlobThatCannotBeGotTest extends CamelTestSupport {
+
+ private final Set<String> blobs = new ConcurrentSkipListSet<>();
+ private final Set<String> archived = ConcurrentHashMap.newKeySet();
+ private final CountDownLatch failedPolls = new CountDownLatch(2);
+ private final CountDownLatch skippedReported = new CountDownLatch(1);
+ private final List<String> handled = new CopyOnWriteArrayList<>();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext context = super.createCamelContext();
+ context.getRegistry().bind("client", serviceClient());
+ context.getRegistry().bind("countPolls", new
PollingConsumerPollStrategy() {
+ @Override
+ public boolean begin(Consumer consumer, Endpoint endpoint) {
+ return true;
+ }
+
+ @Override
+ public void commit(Consumer consumer, Endpoint endpoint, int
polledMessages) {
+ // noop
+ }
+
+ @Override
+ public boolean rollback(Consumer consumer, Endpoint endpoint, int
retryCounter, Exception e) {
+ failedPolls.countDown();
+ return false;
+ }
+ });
+ context.getRegistry().bind("recordErrors", new ExceptionHandler() {
+ @Override
+ public void handleException(Throwable exception) {
+ handled.add(String.valueOf(exception.getMessage()));
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ record(message);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ record(message);
+ }
+
+ private void record(String message) {
+ handled.add(message);
+ if (message.contains("b.txt")) {
+ skippedReported.countDown();
+ }
+ }
+ });
+ return context;
+ }
+
+ @Test
+ void blobThatCannotBeGotDoesNotBlockTheOtherBlobs() throws Exception {
+ blobs.add("a.txt");
+ blobs.add("b.txt");
+ blobs.add("c.txt");
+ archived.add("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedHeaderValuesReceivedInAnyOrder(BlobConstants.BLOB_NAME,
"a.txt", "c.txt");
+
+
addConsumerRoute("&deleteAfterRead=true&exceptionHandler=#recordErrors");
+
+ mock.assertIsSatisfied();
+ assertTrue(skippedReported.await(20, TimeUnit.SECONDS), "the skipped
blob is reported: " + handled);
+ assertTrue(blobs.contains("b.txt"), "b.txt stays");
+ }
+
+ @Test
+ void pollFailsWhenNoListedBlobCanBeGot() throws Exception {
+ blobs.add("a.txt");
+ blobs.add("b.txt");
+ archived.add("a.txt");
+ archived.add("b.txt");
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(0);
+
+ addConsumerRoute("&deleteAfterRead=true&pollStrategy=#countPolls");
+
+ // as for a failed listing (expired credentials, no permission to read
blobs): the poll fails, so the error
+ // handling and the backoff of the scheduled consumer apply
+ assertTrue(failedPolls.await(20, TimeUnit.SECONDS), "the polls fail");
+ mock.assertIsSatisfied();
+ }
+
+ private void addConsumerRoute(String options) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("azure-storage-blob://account/container?serviceClient=#client&initialDelay=0&delay=10"
+ options)
+ .to("mock:result");
+ }
+ });
+ }
+
+ @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 -> blobs.stream().map(name -> new
BlobItem().setName(name)).toList().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 (archived.contains(name)) {
+ // what Azure answers to a read of a blob in the archive tier
+ HttpResponse response = mock(HttpResponse.class);
+ when(response.getStatusCode()).thenReturn(409);
+ throw new BlobStorageException("This operation is not
permitted on an archived blob.", response, null);
+ }
+ BlobInputStream stream = mock(BlobInputStream.class);
+
when(stream.getProperties()).thenReturn(mock(BlobProperties.class));
+ return stream;
+ });
+ doAnswer(i -> {
+ blobs.remove(name);
+ return null;
+ }).when(blobClient).delete();
+ return blobClient;
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 2a293ca9a328..9bfba02ca349 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1619,6 +1619,15 @@ go through one of those paths, remove the `{` and `}`
characters from it in the
endpoint (Chunk has no escape for tags inside data), for example with a
processor that rewrites the body or
header.
+=== camel-azure-storage-blob - a listed blob that cannot be got is skipped
+
+When the consumer lists the container, a failure to get one of the listed
blobs (for example a blob in the archive
+tier, or access denied) no longer fails the poll, as long as another listed
blob could be got: the blob is skipped, the
+other listed blobs are consumed, and the skipped blob is tried again by a
later poll. The failure is given to the
+exception handler of the consumer, with the blob name in the message (by
default a `WARN` with the stack trace, at every
+poll that lists the blob). Before, the poll failed, and as every poll lists
the container again, one such blob stopped
+the consumer from consuming any blob. The poll still fails when none of the
listed blobs could be got.
+
=== camel-azure-servicebus - the dead-letter reason and description are bounded
When the consumer dead-letters a message after a failed route with
`enableDeadLettering=true`, the reason and the
