= Azure Storage Datalake Service Component
:doctitle: Azure Storage Datalake Service
:shortname: azure-storage-datalake
:artifactid: camel-azure-storage-datalake
:description: Camel Azure Datalake Gen2 Component
:since: 3.8
:supportlevel: Stable
:component-header: Both producer and consumer are supported
//Manually maintained attributes
:group: Azure
:camel-spring-boot-name: azure-storage-datalake

*Since Camel {since}*

*{component-header}*

The Azure storage datalake component is used for storing and retrieving file from Azure Storage Datalake Sevice using the *Azure APIs v12*.

Prerequisites

You need to have a valid Azure account with Azure storage set up. More information can be found at https://docs.microsoft.com/azure/[Azure Documentation Portal].

Maven users will need to add the following dependency to their `pom.xml` for this component.

[source,xml]
---------------
<dependency>
    <groupId>org.apache.camel</groupId>
    <artifactId>camel-azure-storage-datalake</artifactId>
    <version>x.x.x</version>
    <!-- use the same version as your camel core version -->
</dependency>
---------------

== Uri Format

[source,text]
----
azure-storage-datalake:accountName[/fileSystemName][?options]
----

In case of a consumer, both accountName and fileSystemName are required. In case of the producer, it depends on the operation
being requested.

You can append query options to the URI in the following format, ?option1=value&option2=value&...


// component-configure options: START

// component-configure options: END

// component options: START
include::partial$component-configure-options.adoc[]
include::partial$component-endpoint-options.adoc[]
// component options: END

// endpoint options: START

// endpoint options: END


=== Methods of authentication
In order to use this component, you will have to provide at least one of the below given points for authentication purposes.

- Provide `accountName` and `accessKey` for your azure account.
- Provide StorageSharedKeyCredential instance which can be provided into `sharedKeyCredential` option.
- Provide ClientSecretCredential instance which can be provided into `clientSecretCredential` option.
- Provide `accountName`, `clientId`, `clientSecret` and `tenantId` for authentication with Azure Active Directory.
- Provide a DataLakeServiceClient instance which can be provided into `serviceClient` option.


== Usage

For example, in order to download content from file `test.txt` located on the `filesystem` in `camelTesting` storage account, use the following snippet:

[source,java]
----
from("azure-storage-datalake:camelTesting/filesystem?fileName=test.txt&accountKey=key").
to("file://fileDirectory");
----

=== Message Headers specified by user

[width="100%",cols="10%,10%,10%,70%",options="header",]
|===
|Header |Variable Name |Type |Description
|`CamelAzureStorageDataLakeListFileSystemsOptions` |`DataLakeConstants.LIST_FILESYSTEMS_OPTIONS` | `ListFileSystemsOptions` | Defines options available to configure the behavior of a call to listFileSystemsSegment on a DataLakeServiceAsyncClient object. Null may be passed.
|`CamelAzureStorageDataLakeTimeout` |`DataLakeConstants.TIMEOUT` |`Duration` | An optional timeout value beyond which a RuntimeException will be raised.
|`CamelAzureStorageDataLakeOperation` |`DataLakeConstants.DATALAKE_OPERATION` |`DataLakeOperationsDefinition` | Specify the producer operation to execute. Different operations allowed are shown below.
|`CamelAzureStorageDataLakeFileSystemName` |`DataLakeConstants.FILESYSTEM_NAME` |`String` | Name of the file system in azure datalake on which operation is to be performed. Please make sure that filesystem name is all lowercase.
|`CamelAzureStorageDataLakeDirectoryName` |`DataLakeConstants.DIRECTORY_NAME` |`String` | Name of the directory in azure datalake on which operation is to be performed.
|`CamelAzureStorageDataLakeFileName` |`DataLakeConstants.FILE_NAME` |`String` | Name of the file in azure datalake on which operation is to be performed.
|`CamelAzureStorageDataLakeMetadata` |`DataLakeConstants.METADATA` |`Map<String,String>` | The metadata to associate with the file.
|`CamelAzureStorageDataLakePublicAccessType` |`DataLakeConstants.PUBLIC_ACCESS_TYPE` |`PublicAccessType` |
|`CamelAzureStorageDataLakeRequestCondition` |`DataLakeConstants.DATALAKE_REQUEST_CONDITION` |`DataLakeRequestConditions` | This contains values which will restrict the successful operation of a variety of requests to the conditions present. These conditions are entirely optional.
|`CamelAzureStorageDataLakeListPathOptions` |`DataLakeConstants.LIST_PATH_OPTIONS` |`ListPathOptions` | Defines options available to configure the behavior of a call to listContainersSegment on a DataLakeFileSystemClient object. Null may be passed.
|`CamelAzureStorageDataLakePath` |`DataLakeConstants.PATH` |`String` | Path of the file to be used for upload operations.
|`CamelAzureStorageDataLakeRecursive` |`DataLakeConstants.RECURSIVE` |`Boolean` | Specifies if the call to listContainersSegment should recursively include all paths.
|`CamelAzureStorageDataLakeMaxResults` |`DataLakeConstants.MAX_RESULTS` |`Integer` | Specifies the maximum number of blobs to return, including all BlobPrefix elements.
|`CamelAzureStorageDataLakeUserPrincipalNameReturned` |`DataLakeConstants.USER_PRINCIPAL_NAME_RETURNED` |`Boolean` | 
|`CamelAzureStorageDataLakeRegex` |`DataLakeConstants.REGEX` |`String` | Filter the results to return only those files with match the specified regular expression.
|`CamelAzureStorageDataLakeFileDir` |`DataLakeConstants.FILE_DIR` |`String` | Directory in which the file is to be downloaded.
|`CamelAzureStorageDataLakeContentMD5` |`DataLakeConstants.CONTENT_MD5` |`byte[]` | An MD5 hash of the content. The hash is used to verify the intergity of the file during transport.
|`CamelAzureStorageDataLakeFileRange` |`DataLakeConstants.FILE_RANGE` |`FileRange` | This is a representation of a range of bytes on a file, typically used during a download operation. Passing null as a FileRange value will default to the entire range of the file.
|`CamelAzureStorageDataLakeParallelTransferOptions` |`DataLakeConstants.PARALLEL_TRANSFER_OPTIONS` | `ParallelTransferOptions` | 
|`CamelAzureStorageDataLakeOpenOptions` |`DataLakeConstants.OPEN_OPTIONS` |`Set<OpenOption>` | Set of OpenOption used to configure how to open or create a file.
|`CamelAzureStorageDataLakeIncrementalCopy` |`DataLakeConstants.INCREMENTAL_COPY` |`Boolean` | Flag indicating if the file was incrementally copied.
|`CamelAzureStorageDataLakeDownloadLinkExpiration` |`DataLakeConstants.DOWNLOAD_LINK_EXPIRATION` |`Long` | Set the Expiration time of the download link.
|`CamelAzureStorageDataLakeFileOffset` |`DataLakeConstants.FILE_OFFSET` |`Long` | The position where the data is to be appended.
|`CamelAzureStorageDataLakeLeaseId` |`DataLakeConstants.LEASE_ID` |`String` | By setting lease id, requests will fail if the provided lease does not match the active lease on the file.
|`CamelAzureStorageDataLakePathHttpHeaders` |`DataLakeConstants.PATH_HTTP_HEADERS` |`PathHttpHeaders` | Additional parameters for a set of operations.
|`CamelAzureStorageDataLakeRetainCommitedData` |`DataLakeConstants.RETAIN_UNCOMMITED_DATA` |`Boolean` | Determines Whether or not uncommitted data is to be retained after the operation.
|`CamelAzureStorageDataLakeClose` |`DataLakeConstants.CLOSE` |`Boolean` | Whether or not a file changed event raised indicates completion (true) or modification (false).
|`CamelAzureStorageDataLakePosition` |`DataLakeConstants.POSITION` |`Long` | The length of the file after all data has been written.
|`CamelAzureStorageDataLakeExpression` |`DataLakeConstants.EXPRESSION` |`String` | The query expression on the file.
|`CamelAzureStorageDataLakeInputSerialization` |`DataLakeConstants.INPUT_SERIALIZATION` |`FileQuerySerialization` | Defines the input serialization for a file query request. either FileQueryJsonSerialization or FileQueryDelimitedSerialization
|`CamelAzureStorageDataLakeOutputSerialization` |`DataLakeConstants.OUTPUT_SERIALIZATION` |`FileQuerySerialization` | Defines the output serialization for a file query request. either FileQueryJsonSerialization or FileQueryDelimitedSerialization
|`CamelAzureStorageDataLakeErrorConsumer` |`DataLakeConstants.ERROR_CONSUMER` |`Consumer<FileQueryError>` | Sets error consumer for file query
|`CamelAzureStorageDataLakeProgressConsumer` |`DataLakeConstants.PROGRESS_CONSUMER` |`Consumer<FileQueryProgress>` | Sets progress consumer for file query
|`CamelAzureStorageDataLakeQueryOptions` |`DataLakeConstants.QUERY_OPTIONS` |`FileQueryOptions` | Optional parameters for File Query.
|`CamelAzureStorageDataLakePermission` |`DataLakeConstants.PERMISSION` |`String` | Sets the permission for file.
|`CamelAzureStorageDataLakeUmask` |`DataLakeConstants.UMASK` |`String` | Sets the umask for file.
|===

=== Message headers set by the component

[width="100%",cols="10%,10%,10%,70%",options="header",]
|===
|Header |Variable Name |Type |Description
|`CamelAzureStorageDataLakeMetadata` |`DataLakeConstants.METADATA` |`Map<String,String>` | The metadata to associate with the file.
|`CamelAzureStorageDataLakeRawHttpHeaders` |`DataLakeConstants.RAW_HTTP_HEADERS` |`HttpHeaders` | Non parsed http headers that can be used by the user.
|`CamelAzureStorageDataLakeAccessTier` |`DataLakeConstants.ACCESS_TIER` |`AccessTier` | Access tier of file.
|`CamelAzureStorageDataLakeAccessTierChangeTime` |`DataLakeConstants.ACCESS_TIER_CHANGE_TIME` |`OffsetDateTime` | Datetime when the access tier of the blob last changed.
|`CamelAzureStorageDataLakeArchiveStatus` |`DataLakeConstants.ARCHIVE_STATUS` |`ArchiveStatus` | Archive status of file. 
|`CamelAzureStorageDataLakeCacheControl` |`DataLakeConstants.CACHE_CONTROL` |`String` | Cache control specified for the file. 
|`CamelAzureStorageDataLakeContentDisposition` |`DataLakeConstants.CONTENT_DISPOSITION` |`String` | Content disposition specified for the file.
|`CamelAzureStorageDataLakeContentEncoding` |`DataLakeConstants.CONTENT_ENCODING` |`String` | Content encoding specified for the file.
|`CamelAzureStorageDataLakeContentLanguage` |`DataLakeConstants.CONTENT_LANGUAGE` |`String` | Content language specified for the file.
|`CamelAzureStorageDataLakeContentType` |`DataLakeConstants.CONTENT_TYPE` |`String` | Content type specified for the file.
|`CamelAzureStorageDataLakeCopyCompletionTime` |`DataLakeConstants.COPY_COMPLETION_TIME` |`OffsetDateTime` | Conclusion time of the last attempted Copy Blob operation where this file was the destination file.
|`CamelAzureStorageDataLakeCopyId` |`DataLakeConstants.COPY_ID` |`String` | String identifier for this copy operation.
|`CamelAzureStorageDataLakeCopyProgress` |`DataLakeConstants.COPY_PROGRESS` |`String` | Contains the number of bytes copied and the total bytes in the source in the last attempted Copy Blob operation where this file was the destination file.
|`CamelAzureStorageDataLakeCopySource` |`DataLakeConstants.COPY_SOURCE` |`String` | URL up to 2 KB in length that specifies the source file or file used in the last attempted Copy Blob operation where this file was the destination file.
|`CamelAzureStorageDataLakeCopyStatus` |`DataLakeConstants.COPY_STATUS` |`CopyStatusType` | Status of the last copy operation performed on the file.
|`CamelAzureStorageDataLakeCopyStatusDescription` |`DataLakeConstants.COPY_STATUS_DESCRIPTION` |`String` |
|`CamelAzureStorageDataLakeCreationTime` |`DataLakeConstants.CREATION_TIME` |`OffsetDateTime` | Creation time of the file.
|`CamelAzureStorageDataLakeEncryptionKeySha256` |`DataLakeConstants.ENCRYPTION_KEY_SHA_256` |`String` | The SHA-256 hash of the encryption key used to encrypt the file.
|`CamelAzureStorageDataLakeServerEncrypted` |`DataLakeConstants.SERVER_ENCRYPTED` |`Boolean` | Flag indicating if the file's content is encrypted on the server.
|`CamelAzureStorageDataLakeETag` |`DataLakeConstants.E_TAG` |`String` | The E Tag of the file.
|`CamelAzureStorageDataLakeFileSize` |`DataLakeConstants.FILE_SIZE` |`Long` | Size of the file.
|`CamelAzureStorageDataLakeLastModified` |`DataLakeConstants.LAST_MODIFIED` |`OffsetDateTime` | Datetime when the file was last modified.
|`CamelAzureStorageDataLakeLeaseDuration` |`DataLakeConstants.LEASE_DURATION` |`LeaseDurationType` | Type of lease on the file.
|`CamelAzureStorageDataLakeLeaseState` |`DataLakeConstants.LEASE_STATE` |`LeaseStateType` | State of the lease on the file.
|`CamelAzureStorageDataLakeLeaseStatus` |`DataLakeConstants.LEASE_STATUS` |`LeaseStatusType` | Status of the lease on the file.
|`CamelAzureStorageDataLakeDownloadLink` |`DataLakeConstants.DOWNLOAD_LINK` |`String` | The link that can be used to download the file from datalake.
|===

=== Automatic detection of service client

The component is capable of automatically detecting the presence of a DataLakeServiceClient bean in the registry.
Hence, if your registry has only one instance of type DataLakeServiceClient, it will be automatically used as the default client.
You won't have to explicitly define it as an uri paramater.

=== Azure Storage DataLake Producer Operations

The various operations supported by Azure Storage DataLake are as given below:

*Operations on Service level*

For these operations, `accountName` option is required
[width="100%", cols="10%,90%", options="header",]
|===
|Operation |Description
|`listFileSystem` | List all the file systems that are present in the given azure account.
|===

*Operations on File system level*

For these operations, `accountName` and `fileSystemName` options are required
[width="100%", cols="10%,90%", options="header",]
|===
|Operation |Description
|`createFileSystem` | Creates a new file System with the storage account
|`deleteFileSystem` | Deletes the specified file system within the storage account
|`listPaths` | Returns list of all the files within the given path in the given file system , with folder structure flattened
|===

*Operations on Directory level*

For these operations, `accountName`, `fileSystemName` and `directoryName` options are required
[width="100%", cols="10%,90%", options="header",]
|===
|Operation |Description
|`createFile` | Creates a new file in the specified directory within the fileSystem
|`deleteDirectory` | Deletes the specified directory within the file system
|===

*Operations on file level*

For these operations, `accountName`, `fileSystemName` and `fileName` options are required
[width="100%", cols="10%,90%", options="header",]
|===
|Operation |Description
|`getFile` | Get the contents of a file
|`downloadToFile` | Downloadd the entire file from the file system into a path specified by fileDir.
|`downloadLink` | Generate download link for the specified file using Shared Access Signature (SAS).
The expiration time to be set for the link can be specified otherwise 1 hour is taken as default.
|`deleteFile` | Deletes the specified file.
|`appendToFile` | Appends the data passed to the specified file in the file System. Flush command is
required after append.
|`flushToFile` | Flushes the data already appended to the specified file.
|`openQueryInputStream` | Opens an inputstream based on the query passed to the endpoint. For this operation,
you must first register the query acceleration feature with your subscription.
|===

Refer the examples section below for more details on how to use these operations

=== Consumer Examples
To consume a file from the storage datalake into a file using the file component, this can be done like this:

[source,java]
----
from("azure-storage-datalake":cameltesting/filesystem?fileName=test.txt&accountKey=yourAccountKey").
to("file:/filelocation");
----

You can also directly write to a file without using the file component. For this, you will need to specify the path in `fileDir` option, to save it to your machine.

[source,java]
----
from("azure-storage-datalake":cameltesting/filesystem?fileName=test.txt&accountKey=yourAccountKey&fileDir=/test/directory").
to("mock:results");
----

This component also supports batch consumer. So, you can consume multiple files from a file system by specifying the path
from where you want to consume the files.

[source,java]
----
from("azure-storage-datalake":cameltesting/filesystem?accountKey=yourAccountKey&fileDir=/test/directory&path=abc/test").
to("mock:results");
----


=== Producer Examples
-  `listFileSystem`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        //required headers can be added here
        exchange.getIn().setHeader(DataLakeConstants.LIST_FILESYSTEMS_OPTIONS, new ListFileSystemsOptions().setMaxResultsPerPage(10));
    })
    .to("azure-storage-datalake:cameltesting?operation=listFileSystem&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `createFileSystem`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.FILESYSTEM_NAME, "test1");
    })
    .to("azure-storage-datalake:cameltesting?operation=createFileSystem&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `deleteFileSystem`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.FILESYSTEM_NAME, "test1");
    })
    .to("azure-storage-datalake:cameltesting?operation=deleteFileSystem&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `listPaths`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.LIST_PATH_OPTIONS, new ListPathsOptions().setPath("/main"));
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=listPaths&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `getFile`

This can be done in two ways, We can either set an outputstream in the exchange body

[source,java]
----
from("direct:start")
    .process(exchange -> {
        // set an outputstream where the file data can should be written
        exchange.getIn().setBody(outputStream);
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=getFile&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

Or if body is not set, the operation will give an inputstream, given that you have already registered for query acceleration
in azure portal.

[source,java]
----
from("direct:start")
    .to("azure-storage-datalake:cameltesting/filesystem?operation=getFile&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .process(exchange -> {
        InputStream inputStream = exchange.getMessage().getBody(InputStream.class);
        System.out.Println(IOUtils.toString(inputStream, StandardCharcets.UTF_8.name()));
    })
    .to("mock:results");
----

-  `deleteFile`

[source,java]
----
from("direct:start")
    .to("azure-storage-datalake:cameltesting/filesystem?operation=deleteFile&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

- `downloadToFile`

[source,java]
----
from("direct:start")
    .to("azure-storage-datalake:cameltesting/filesystem?operation=downloadToFile&fileName=test.txt&fileDir=/test/mydir&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `downloadLink`

[source,java]
----
from("direct:start")
    .to("azure-storage-datalake:cameltesting/filesystem?operation=downloadLink&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .process(exchange -> {
        String link = exchange.getMessage().getBody(String.class);
        System.out.println(link);
    })
    .to("mock:results");
----

-  `appendToFile`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        final String data = "test data";
        final InputStream inputStream = new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8));
        exchange.getIn().setBody(inputStream);
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=appendToFile&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `flushToFile`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.POSITION, 0);
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=flushToFile&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `openQueryInputStream`

For this operation, you should have already registered for query acceleration on the azure portal

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.QUERY_OPTIONS, new FileQueryOptions("SELECT * from BlobStorage"));
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=openQueryInputStream&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `upload`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        final String data = "test data";
        final InputStream inputStream = new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8));
        exchange.getIn().setBody(inputStream);
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=upload&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `uploadFromFile`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.PATH, "test/file.txt");
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=uploadFromFile&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `createFile`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.DIRECTORY_NAME, "test/file/");
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=createFile&fileName=test.txt&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

-  `deleteDirectory`

[source,java]
----
from("direct:start")
    .process(exchange -> {
        exchange.getIn().setHeader(DataLakeConstants.DIRECTORY_NAME, "test/file/");
    })
    .to("azure-storage-datalake:cameltesting/filesystem?operation=deleteDirectory&dataLakeServiceClient=#serviceClient")
    .to("mock:results");
----

=== Testing

Please run all the unit tests and integration test while making changes to the component as changes or version upgrades can break things.
For running all the test in the component, you will need to obtain azure accountName and accessKey. After obtaining the same, you
can run the full test, on this component directory, by running the following maven command

[source,bash]
----
mvn verify -PfullTests -Dazure.storage.account.name=<accountName> -Dazure.storage.account.key=<accessKey>
----

You can also skip the integration test, and run only basic unit test by using the command

[source,bash]
----
mvn test
----

include::spring-boot:partial$starter.adoc[]
