Federico Mariani created CAMEL-26655:
----------------------------------------
Summary: camel-azure-storage-blob - consumer lists every page and
opens every blob before routing; add maxMessagesPerPoll
Key: CAMEL-26655
URL: https://issues.apache.org/jira/browse/CAMEL-26655
Project: Camel
Issue Type: Improvement
Components: camel-azure
Affects Versions: 4.18.5, 4.23.0, 4.22.2
Reporter: Federico Mariani
h3. Problem
At every poll (default {{delay=500}}), the consumer without {{blobName}} does
the following, on main at commit 7978db3d42dd:
* It reads the whole container listing.
{{components/camel-azure/camel-azure-storage-blob/src/main/java/org/apache/camel/component/azure/storage/blob/client/BlobContainerClientWrapper.java}}
line 50 calls {{client.listBlobs(options, timeout).stream().toList()}}, which
fetches every page of the SDK's lazy {{PagedIterable}}: one List Blobs call per
page of up to 5000 blobs.
* It applies {{regex}} on the client after the full listing
({{BlobContainerOperations.java}} lines 54-62). When {{regex}} is set,
{{prefix}} is not sent ({{BlobConfigurationOptionsProxy.java}} lines 72-78).
* It gets every listed blob before routing the first
({{BlobConsumer.createBatchExchangesFromContainer}}, {{BlobConsumer.java}}
lines 129-175, loop at 145-154):
** without {{fileDir}}, {{getBlob}} calls {{BlobClient.openInputStream}}. In
azure-storage-blob 12.32.0 (the version Camel uses), that call downloads the
first chunk (4 MiB by default) into memory and blocks ({{BlobClientBase.java}}
lines 509-519 and 564). Read from source.
** with {{fileDir}}, every blob is downloaded in full to disk.
* There is no {{maxMessagesPerPoll}} option. The description of
{{maxResultsPerPage}} reads "Specifies the maximum number of blobs to return,
including all BlobPrefix elements", but the value is only the page size
({{BlobConfigurationOptionsProxy.java}} lines 97-100).
With {{deleteAfterRead=false}} (the default), re-delivering every blob at every
poll is by design and documented. The {{deleteAfterRead}} javadoc and the
consumer page say to use the Idempotent Consumer EIP. That remedy runs only
after the consumer has got the blob, so every poll still lists every page and
opens every blob. With Camel's default stream caching, each routed blob is also
read in full at route entry, before the EIP drops it.
h3. How to reproduce
Setup (measured on main): the real azure-storage-blob 12.32.0 client (Netty)
talks to an embedded fake Blob service (JDK {{HttpServer}}). The fake serves 3
pages x 4 blobs of 5 MiB each, honours {{marker}}, {{maxresults}} and
{{x-ms-range}}, and counts requests and bytes. The counters are read in the
route when the first exchange of the first poll arrives. The route is
{{azure-storage-blob://account/container?serviceClient=#client&maxResultsPerPage=4}};
the fake saw {{maxresults=4}} on every List Blobs call.
||Route||List Blobs calls||Get Blob calls||Bytes sent before the first
exchange||BATCH_SIZE||
|{{noStreamCaching()}}|3|12, all {{x-ms-range: bytes=0-4194303}}|48 MiB|12|
|default (stream caching on)|3|13 (the 13th reads the rest of the first
blob)|49 MiB|12|
|{{fileDir}} set|3|24|60 MiB (every blob in full)|12|
|{{regex}} matching only the first blob|3|2|5 MiB|1|
Re-delivery: with {{delay=50}} and
{{.idempotentConsumer(header("CamelAzureStorageBlobBlobName"),
memoryIdempotentRepository())}}, 3 polls produced 9 List Blobs calls, 36 blob
opens, 60 Get Blob calls and 168 MiB transferred, while only the 12 first-poll
exchanges passed the EIP. Adding {{maxMessagesPerPoll=2}} to the URI fails with
{{ResolveEndpointFailedException}}.
Estimate, from the source and not measured: per poll, ceil(N/5000) List Blobs
calls plus N sequential Get Blob calls, and up to sum(min(size, 4 MiB))
buffered before the first exchange is routed.
h3. Proposed change
* *List lazily and get just in time.* Iterate the {{PagedIterable}} instead of
{{toList()}}, and get each blob (open the stream, or download it to
{{fileDir}}) in {{processBatch}} right before routing it. The minio consumer
made the same change in CAMEL-17100 (7de327d91eb6).
* *Add {{maxMessagesPerPoll}}.* Stop listing once M blobs have been *got*.
Count got blobs, not listed ones: since CAMEL-25770, a blob that cannot be got
(archive tier, access denied) is skipped. If the cap counted listed blobs, M
such blobs at the head of the name-ordered listing would starve the consumer,
which is the CAMEL-25770 symptom again. Blobs skipped as in progress
(CAMEL-25551) or gone (404) must not count either.
* *Keep the CAMEL-25770 rule.* The poll fails only when no listed blob could be
got. With just-in-time gets this still works, because nothing has been routed
before the first success.
* *Fix the batch properties.* {{BATCH_SIZE}} and {{BATCH_COMPLETE}} are set
from the queue size before routing ({{BlobConsumer.java}} lines 249-251). If
blobs are skipped at get time, the last exchange may never get
{{BATCH_COMPLETE=true}}. Compute them after the get, or fetch one blob ahead.
* *Avoid starvation without delete or move.* With a cap and no
{{deleteAfterRead}}/{{moveAfterRead}}, every poll restarts at the first blob,
so blobs after the first M are never reached. Keep the listing marker or the
last blob name across polls, and wrap around when the listing is exhausted.
* *Document the cost.* The docs should say that the Idempotent Consumer EIP
does not save the downloads. Optionally, add a consumer-side repository so that
blobs already consumed are skipped before {{getBlob}}.
h3. Impact / upgrade
* Just-in-time gets change when a blob is read: at routing time instead of at
listing time. A non-zero default for {{maxMessagesPerPoll}} would be a changed
default. Both need an entry in the 4.24 upgrade guide.
* Seen on 4.18.5, 4.22.2 and 4.23.0 (the same listing and get loop).
* The camel-azure-storage-datalake consumer has the same full listing and
up-front open ({{DataLakeConsumer.java}} lines 91-117). It also has no delete
or move option.
h3. Related
* CAMEL-25770 (a9ef7248b7ce): skip a listed blob that cannot be got.
* CAMEL-25551 (ab5932958d36): the in-progress set.
* CAMEL-16826 (86c1b61d7f9f): {{deleteAfterRead}}, default false.
* CAMEL-25758: {{moveAfterRead}}.
* CAMEL-17100 (7de327d91eb6): minio, the same change.
* CAMEL-26635: the file family, stop listing at {{maxMessagesPerPoll}}.
_Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
