turcsanyip commented on a change in pull request #4126: NIFI-7103 Adding PutAzureDataLakeStorage Processor to provide native support for Azure Data Lake Storage Gen 2 URL: https://github.com/apache/nifi/pull/4126#discussion_r400524840
########## File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/PutAzureDataLakeStorage.java ########## @@ -0,0 +1,96 @@ +/* + * 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.nifi.processors.azure.storage; + +import java.io.BufferedInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.WritesAttribute; +import org.apache.nifi.annotation.behavior.WritesAttributes; +import org.apache.nifi.annotation.documentation.CapabilityDescription; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.flowfile.FlowFile; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.ProcessSession; +import org.apache.nifi.processor.exception.ProcessException; + +import com.azure.storage.file.datalake.DataLakeDirectoryClient; +import com.azure.storage.file.datalake.DataLakeFileClient; +import com.azure.storage.file.datalake.DataLakeFileSystemClient; +import com.azure.storage.file.datalake.DataLakeServiceClient; +import org.apache.nifi.processors.azure.AbstractAzureDataLakeStorageProcessor; +import com.azure.storage.file.datalake.implementation.models.StorageErrorException; + +@Tags({"azure", "microsoft", "cloud", "storage", "adlsgen2", "datalake"}) +@CapabilityDescription("Puts content into an Azure Data Lake Storage Gen 2") +@WritesAttributes({@WritesAttribute(attribute = "azure.filesystem", description = "The name of the Azure File System"), + @WritesAttribute(attribute = "azure.directory", description = "The name of the Azure Directory"), + @WritesAttribute(attribute = "azure.filename", description = "The name of the Azure File Name"), + @WritesAttribute(attribute = "azure.primaryUri", description = "Primary location for file content"), + @WritesAttribute(attribute = "azure.length", description = "Length of the file")}) +@InputRequirement(Requirement.INPUT_REQUIRED) + +public class PutAzureDataLakeStorage extends AbstractAzureDataLakeStorageProcessor { + + @Override + public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException { + FlowFile flowFile = session.get(); + if (flowFile == null) { + return; + } + final long startNanos = System.nanoTime(); + try { + final String fileSystem = context.getProperty(FILESYSTEM).evaluateAttributeExpressions(flowFile).getValue(); + final String directory = context.getProperty(DIRECTORY).evaluateAttributeExpressions(flowFile).getValue(); + final String fileName = context.getProperty(FILE).evaluateAttributeExpressions(flowFile).getValue(); + final DataLakeServiceClient storageClient = getStorageClient(context, flowFile); + final DataLakeFileSystemClient dataLakeFileSystemClient = storageClient.getFileSystemClient(fileSystem); + final DataLakeDirectoryClient directoryClient = dataLakeFileSystemClient.getDirectoryClient(directory); + final DataLakeFileClient fileClient = directoryClient.createFile(fileName); + final long length = flowFile.getSize(); + if (length > 0) { + try (final InputStream rawIn = session.read(flowFile); final BufferedInputStream in = new BufferedInputStream(rawIn)) { + fileClient.append(in, 0, length); + + } + } + fileClient.flush(length); + final Map<String, String> attributes = new HashMap<>(); + attributes.put("azure.filesystem", fileSystem); + attributes.put("azure.directory", directory); + attributes.put("azure.filename", fileName); + attributes.put("azure.primaryUri", fileClient.getFileUrl()); + attributes.put("azure.length", String.valueOf(length)); + flowFile = session.putAllAttributes(flowFile, attributes); + + + session.transfer(flowFile, REL_SUCCESS); + final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + session.getProvenanceReporter().send(flowFile, fileClient.getFileUrl(), transferMillis); + } catch (IOException | StorageErrorException | IllegalArgumentException e) { Review comment: I'd recommend to catch `Exception` here. I managed to throw `DataLakeStorageException` from the processor (eg. via specifying a non-existing filesystem or an invalid directory name on the UI). In this case the flowfile does not go to failure, but always goes back to the input queue of the processor which is not the expected behaviour I think (a wrong flowfile could hang up the processor, the other flowfiles would not be processed). `DataLakeStorageException` is just an example I think so I would not add it as a fourth exception to be caught but rather catch `Exception`. Or catch `(IOException | AzureException | IllegalArgumentException)`, but I'm afraid there can be other non-expected `RuntimeException-s` coming out from the library, so `Exception` seems to me the safest solution. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
