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 9d34c02e4dea CAMEL-26583: camel-azure-storage-datalake - apply ranges
to streaming downloads (#29156)
9d34c02e4dea is described below
commit 9d34c02e4dea4c63a4a8d00781325fd458af9f4f
Author: Andrea Cosentino <[email protected]>
AuthorDate: Wed Nov 4 19:53:40 2026 +0100
CAMEL-26583: camel-azure-storage-datalake - apply ranges to streaming
downloads (#29156)
Streaming downloads of the camel-azure-storage-datalake getFile operation
now honour the requested file range and request conditions, instead of always
reading the whole file. Upgrade guide and docs updated.
AI-generated by Codex via OSS Helper on behalf of @oscerd.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
Co-authored-by: OpenAI Codex <[email protected]>
---
.../docs/azure-storage-datalake-component.adoc | 8 ++
.../docs/azure-storage-datalake-component.adoc | 8 ++
.../datalake/client/DataLakeFileClientWrapper.java | 7 +
.../operations/DataLakeFileOperations.java | 6 +-
.../operations/DataLakeFileOperationTest.java | 150 +++++++++++++++------
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 11 ++
6 files changed, 150 insertions(+), 40 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-datalake-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-datalake-component.adoc
index 526f4fbcbad5..a21964ee4631 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-datalake-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/azure-storage-datalake-component.adoc
@@ -414,6 +414,14 @@ from("direct:start")
- `getFile`
+The `CamelAzureStorageDataLakeFileRange` header selects the byte range to
read. Otherwise, set both
+`fileOffset` and `dataCount` on the endpoint to select a range. This applies
whether the operation
+returns an `InputStream` or writes to an `OutputStream` supplied as the
message body.
+The `CamelAzureStorageDataLakeRequestCondition` header applies request
conditions, such as an ETag
+or lease, to both download paths, including streaming reads without a range.
+Consumers without `fileDir` also use the streaming path and honor the endpoint
`fileOffset` and
+`dataCount` options. Remove these options to consume the entire file.
+
This can be done in two ways. We can either set an `OutputStream` in the
exchange body:
._Java-only: sets an `OutputStream` for the file content to be written to_
diff --git
a/components/camel-azure/camel-azure-storage-datalake/src/main/docs/azure-storage-datalake-component.adoc
b/components/camel-azure/camel-azure-storage-datalake/src/main/docs/azure-storage-datalake-component.adoc
index 526f4fbcbad5..a21964ee4631 100644
---
a/components/camel-azure/camel-azure-storage-datalake/src/main/docs/azure-storage-datalake-component.adoc
+++
b/components/camel-azure/camel-azure-storage-datalake/src/main/docs/azure-storage-datalake-component.adoc
@@ -414,6 +414,14 @@ from("direct:start")
- `getFile`
+The `CamelAzureStorageDataLakeFileRange` header selects the byte range to
read. Otherwise, set both
+`fileOffset` and `dataCount` on the endpoint to select a range. This applies
whether the operation
+returns an `InputStream` or writes to an `OutputStream` supplied as the
message body.
+The `CamelAzureStorageDataLakeRequestCondition` header applies request
conditions, such as an ETag
+or lease, to both download paths, including streaming reads without a range.
+Consumers without `fileDir` also use the streaming path and honor the endpoint
`fileOffset` and
+`dataCount` options. Remove these options to consume the entire file.
+
This can be done in two ways. We can either set an `OutputStream` in the
exchange body:
._Java-only: sets an `OutputStream` for the file content to be written to_
diff --git
a/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java
b/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java
index d5f888597a39..99b15d07931b 100644
---
a/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java
+++
b/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/client/DataLakeFileClientWrapper.java
@@ -35,6 +35,7 @@ import com.azure.storage.file.datalake.models.PathHttpHeaders;
import com.azure.storage.file.datalake.models.PathInfo;
import com.azure.storage.file.datalake.models.PathProperties;
import com.azure.storage.file.datalake.options.DataLakeFileAppendOptions;
+import com.azure.storage.file.datalake.options.DataLakeFileInputStreamOptions;
import com.azure.storage.file.datalake.options.FileParallelUploadOptions;
import com.azure.storage.file.datalake.options.FileQueryOptions;
import com.azure.storage.file.datalake.sas.DataLakeServiceSasSignatureValues;
@@ -66,6 +67,12 @@ public class DataLakeFileClientWrapper {
return client.openInputStream().getInputStream();
}
+ public InputStream openInputStream(final FileRange range, final
DataLakeRequestConditions requestConditions) {
+ return client
+ .openInputStream(new
DataLakeFileInputStreamOptions().setRange(range).setRequestConditions(requestConditions))
+ .getInputStream();
+ }
+
public Response<InputStream> openQueryInputStreamWithResponse(final
FileQueryOptions queryOptions) {
return client.openQueryInputStreamWithResponse(queryOptions);
}
diff --git
a/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperations.java
b/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperations.java
index 6f4c13984fe9..117e18a85815 100644
---
a/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperations.java
+++
b/components/camel-azure/camel-azure-storage-datalake/src/main/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperations.java
@@ -73,12 +73,14 @@ public class DataLakeFileOperations {
}
final DataLakeFileClientWrapper fileClientWrapper =
getFileClientWrapper(exchange);
+ final FileRange fileRange = configurationProxy.getFileRange(exchange);
if (outputStream == null) {
- InputStream fileInputStream = fileClientWrapper.openInputStream();
+ final DataLakeRequestConditions requestConditions =
configurationProxy.getDataLakeRequestConditions(exchange);
+ InputStream fileInputStream = fileRange == null &&
requestConditions == null
+ ? fileClientWrapper.openInputStream() :
fileClientWrapper.openInputStream(fileRange, requestConditions);
return new DataLakeOperationResponse(fileInputStream);
}
- final FileRange fileRange = configurationProxy.getFileRange(exchange);
final FileCommonRequestOptions fileCommonRequestOptions =
getCommonRequestOptions(exchange);
final DownloadRetryOptions downloadRetryOptions =
getDownloadRetryOptions(configurationProxy);
diff --git
a/components/camel-azure/camel-azure-storage-datalake/src/test/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperationTest.java
b/components/camel-azure/camel-azure-storage-datalake/src/test/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperationTest.java
index 6aec092b50d3..d41945dd20b7 100644
---
a/components/camel-azure/camel-azure-storage-datalake/src/test/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperationTest.java
+++
b/components/camel-azure/camel-azure-storage-datalake/src/test/java/org/apache/camel/component/azure/storage/datalake/operations/DataLakeFileOperationTest.java
@@ -27,8 +27,12 @@ import java.time.OffsetDateTime;
import com.azure.core.http.HttpHeaders;
import com.azure.core.http.rest.ResponseBase;
import com.azure.storage.file.datalake.DataLakeFileClient;
+import
com.azure.storage.file.datalake.models.DataLakeFileOpenInputStreamResult;
+import com.azure.storage.file.datalake.models.DataLakeRequestConditions;
+import com.azure.storage.file.datalake.models.FileRange;
import com.azure.storage.file.datalake.models.PathInfo;
import com.azure.storage.file.datalake.models.PathProperties;
+import com.azure.storage.file.datalake.options.DataLakeFileInputStreamOptions;
import org.apache.camel.Exchange;
import org.apache.camel.component.azure.storage.datalake.DataLakeConfiguration;
import org.apache.camel.component.azure.storage.datalake.DataLakeConstants;
@@ -39,11 +43,14 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
@@ -97,6 +104,62 @@ public class DataLakeFileOperationTest extends
CamelTestSupport {
assertNotNull(secondResponse.getHeaders());
}
+ @Test
+ void testDownloadToFileRejectsPathTraversalName() {
+ // a file name is influenced by whoever writes to the file system and
may contain ../ segments;
+ // the download must stay within the configured fileDir
+ configuration.setFileDir("/tmp/camel-azure-datalake-download");
+ when(client.getFileName()).thenReturn("../escaped/PROOF_PWNED");
+
+ final DataLakeFileOperations operations = new
DataLakeFileOperations(configuration, client);
+ final Exchange exchange = new DefaultExchange(context);
+
+ assertThrows(IllegalArgumentException.class, () ->
operations.downloadToFile(exchange));
+ }
+
+ @Test
+ void testUploadFile() throws Exception {
+ final OffsetDateTime time = OffsetDateTime.now();
+ final PathInfo pathInfo = new PathInfo("testTag", time);
+ final HttpHeaders httpHeaders = new HttpHeaders();
+ when(client.uploadWithResponse(any(), any()))
+ .thenReturn(new ResponseBase<>(null, 200, httpHeaders,
pathInfo, null));
+
+ final Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody(new
ByteArrayInputStream("testing".getBytes(Charset.defaultCharset())));
+
+ final DataLakeFileOperations operations = new
DataLakeFileOperations(configuration, client);
+ final DataLakeOperationResponse response = operations.upload(exchange);
+
+ assertNotNull(response);
+ assertTrue((boolean) response.getBody());
+ assertNotNull(response.getHeaders());
+ assertEquals("testTag",
response.getHeaders().get(DataLakeConstants.E_TAG));
+ assertEquals(time,
response.getHeaders().get(DataLakeConstants.LAST_MODIFIED));
+ }
+
+ @Test
+ void testServiceClientOverride() throws Exception {
+ final HttpHeaders httpHeaders = new HttpHeaders();
+ final byte[] testing = "testing".getBytes(Charset.defaultCharset());
+ when(alternateClient.appendWithResponse(any(), anyLong(), anyLong(),
any(), any(), any()))
+ .thenReturn(new ResponseBase<>(null, 200, httpHeaders, null,
null));
+ final PathProperties properties = mock(PathProperties.class);
+
when(properties.getFileSize()).thenReturn(Long.valueOf(testing.length));
+ when(alternateClient.getProperties()).thenReturn(properties);
+
+ final Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(DataLakeConstants.FILE_CLIENT,
alternateClient);
+ exchange.getIn().setBody(new ByteArrayInputStream(testing));
+
+ final DataLakeFileOperations operations = new
DataLakeFileOperations(configuration, client);
+ final DataLakeOperationResponse response =
operations.appendToFile(exchange);
+
+ assertNotNull(response);
+ assertTrue((boolean) response.getBody());
+ assertNotNull(response.getHeaders());
+ }
+
@Test
void testFlushUsesAbsoluteHeaderPosition() {
when(client.flushWithResponse(anyLong(), any(), any(), any(), any(),
any()))
@@ -142,58 +205,69 @@ public class DataLakeFileOperationTest extends
CamelTestSupport {
}
@Test
- void testDownloadToFileRejectsPathTraversalName() {
- // a file name is influenced by whoever writes to the file system and
may contain ../ segments;
- // the download must stay within the configured fileDir
- configuration.setFileDir("/tmp/camel-azure-datalake-download");
- when(client.getFileName()).thenReturn("../escaped/PROOF_PWNED");
-
- final DataLakeFileOperations operations = new
DataLakeFileOperations(configuration, client);
+ void testGetFileUsesHeaderRangeForInputStream() throws IOException {
+ configuration.setFileOffset(2L);
+ configuration.setDataCount(3L);
final Exchange exchange = new DefaultExchange(context);
-
- assertThrows(IllegalArgumentException.class, () ->
operations.downloadToFile(exchange));
+ exchange.getIn().setHeader(DataLakeConstants.FILE_RANGE, new
FileRange(4, 2L));
+ assertInputStreamRange(exchange, 4, 2L);
}
@Test
- void testUploadFile() throws Exception {
- final OffsetDateTime time = OffsetDateTime.now();
- final PathInfo pathInfo = new PathInfo("testTag", time);
- final HttpHeaders httpHeaders = new HttpHeaders();
- when(client.uploadWithResponse(any(), any()))
- .thenReturn(new ResponseBase<>(null, 200, httpHeaders,
pathInfo, null));
+ void testGetFileUsesConfiguredRangeForInputStream() throws IOException {
+ configuration.setFileOffset(2L);
+ configuration.setDataCount(3L);
+ assertInputStreamRange(new DefaultExchange(context), 2, 3L);
+ }
+ private void assertInputStreamRange(Exchange exchange, long offset, long
count) throws IOException {
+ final DataLakeFileInputStreamOptions options =
inputStreamOptions(exchange);
+ assertEquals(offset, options.getRange().getOffset());
+ assertEquals(count, options.getRange().getCount());
+ }
+
+ @Test
+ void testGetFileAppliesConditionsWithHeaderRange() throws IOException {
+ configuration.setFileOffset(2L);
+ configuration.setDataCount(3L);
final Exchange exchange = new DefaultExchange(context);
- exchange.getIn().setBody(new
ByteArrayInputStream("testing".getBytes(Charset.defaultCharset())));
+ final DataLakeRequestConditions conditions = new
DataLakeRequestConditions().setIfMatch("etag");
+ exchange.getIn().setHeader(DataLakeConstants.FILE_RANGE, new
FileRange(4, 2L));
+
exchange.getIn().setHeader(DataLakeConstants.DATALAKE_REQUEST_CONDITION,
conditions);
- final DataLakeFileOperations operations = new
DataLakeFileOperations(configuration, client);
- final DataLakeOperationResponse response = operations.upload(exchange);
+ final DataLakeFileInputStreamOptions options =
inputStreamOptions(exchange);
- assertNotNull(response);
- assertTrue((boolean) response.getBody());
- assertNotNull(response.getHeaders());
- assertEquals("testTag",
response.getHeaders().get(DataLakeConstants.E_TAG));
- assertEquals(time,
response.getHeaders().get(DataLakeConstants.LAST_MODIFIED));
+ assertEquals(4, options.getRange().getOffset());
+ assertEquals(2L, options.getRange().getCount());
+ assertSame(conditions, options.getRequestConditions());
}
@Test
- void testServiceClientOverride() throws Exception {
- final HttpHeaders httpHeaders = new HttpHeaders();
- final byte[] testing = "testing".getBytes(Charset.defaultCharset());
- when(alternateClient.appendWithResponse(any(), anyLong(), anyLong(),
any(), any(), any()))
- .thenReturn(new ResponseBase<>(null, 200, httpHeaders, null,
null));
- final PathProperties properties = mock(PathProperties.class);
-
when(properties.getFileSize()).thenReturn(Long.valueOf(testing.length));
- when(alternateClient.getProperties()).thenReturn(properties);
-
+ void testGetFileAppliesConditionsWithoutRange() throws IOException {
final Exchange exchange = new DefaultExchange(context);
+ final DataLakeRequestConditions conditions = new
DataLakeRequestConditions().setLeaseId("lease");
+
exchange.getIn().setHeader(DataLakeConstants.DATALAKE_REQUEST_CONDITION,
conditions);
+
+ final DataLakeFileInputStreamOptions options =
inputStreamOptions(exchange);
+
+ assertNull(options.getRange());
+ assertSame(conditions, options.getRequestConditions());
+ }
+
+ private DataLakeFileInputStreamOptions inputStreamOptions(Exchange
exchange) throws IOException {
+ final InputStream stream = new
ByteArrayInputStream("range".getBytes(Charset.defaultCharset()));
+ final DataLakeFileOpenInputStreamResult result =
mock(DataLakeFileOpenInputStreamResult.class);
+ when(result.getInputStream()).thenReturn(stream);
+
when(alternateClient.openInputStream(any(DataLakeFileInputStreamOptions.class))).thenReturn(result);
exchange.getIn().setHeader(DataLakeConstants.FILE_CLIENT,
alternateClient);
- exchange.getIn().setBody(new ByteArrayInputStream(testing));
- final DataLakeFileOperations operations = new
DataLakeFileOperations(configuration, client);
- final DataLakeOperationResponse response =
operations.appendToFile(exchange);
+ final DataLakeOperationResponse response = new
DataLakeFileOperations(configuration, client).getFile(exchange);
- assertNotNull(response);
- assertTrue((boolean) response.getBody());
- assertNotNull(response.getHeaders());
+ assertEquals(stream, response.getBody());
+ final ArgumentCaptor<DataLakeFileInputStreamOptions> options
+ =
ArgumentCaptor.forClass(DataLakeFileInputStreamOptions.class);
+ verify(alternateClient).openInputStream(options.capture());
+ return options.getValue();
}
+
}
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 b05c3cc968c5..fc8c275c33b7 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
@@ -4654,6 +4654,17 @@ lists the file). A file deleted after the listing is
skipped as well; before, it
Before, any other failure failed the poll, and as every poll lists the file
system again, one such file stopped the
consumer from consuming any file. The poll still fails when none of the listed
files could be got.
+=== camel-azure-storage-datalake - streaming downloads honor ranges and
request conditions
+
+The `getFile` operation now honors the `CamelAzureStorageDataLakeFileRange`
header, or the endpoint
+`fileOffset` and `dataCount` options, when it returns an `InputStream`.
Previously, these settings
+only applied when the message body supplied an `OutputStream`; streaming reads
returned the entire file.
+This also affects consumers without `fileDir`: they now deliver the configured
byte range instead of the whole file.
+Remove the range header and options if your route needs the entire file.
+
+The streaming path also honors the `CamelAzureStorageDataLakeRequestCondition`
header, whether or not a range
+is set. An ETag mismatch or an invalid lease can now fail a streaming read
that previously ignored the condition.
+
=== camel-chatscript - the first message of a conversation is sent, and the
headers are kept
The first message of a ChatScript endpoint, and every message with
`resetChat=true`, used to start the conversation
