turcsanyip commented on a change in pull request #4438: URL: https://github.com/apache/nifi/pull/4438#discussion_r463880311
########## File path: nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureDataLakeStorage.java ########## @@ -0,0 +1,243 @@ +/* + * 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 com.azure.storage.file.datalake.DataLakeFileSystemClient; +import com.azure.storage.file.datalake.DataLakeServiceClient; +import com.azure.storage.file.datalake.models.ListPathsOptions; +import org.apache.commons.lang3.RegExUtils; +import org.apache.commons.lang3.StringUtils; +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.apache.nifi.annotation.behavior.InputRequirement; +import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; +import org.apache.nifi.annotation.behavior.PrimaryNodeOnly; +import org.apache.nifi.annotation.behavior.Stateful; +import org.apache.nifi.annotation.behavior.TriggerSerially; +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.SeeAlso; +import org.apache.nifi.annotation.documentation.Tags; +import org.apache.nifi.annotation.lifecycle.OnScheduled; +import org.apache.nifi.annotation.lifecycle.OnStopped; +import org.apache.nifi.components.PropertyDescriptor; +import org.apache.nifi.components.ValidationContext; +import org.apache.nifi.components.ValidationResult; +import org.apache.nifi.components.state.Scope; +import org.apache.nifi.context.PropertyContext; +import org.apache.nifi.processor.ProcessContext; +import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.util.StandardValidators; +import org.apache.nifi.processor.util.list.AbstractListProcessor; +import org.apache.nifi.processors.azure.storage.utils.ADLSFileInfo; +import org.apache.nifi.serialization.record.RecordSchema; + +import java.io.IOException; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; +import java.util.stream.Collectors; + +import static org.apache.nifi.processor.util.list.ListedEntityTracker.INITIAL_LISTING_TARGET; +import static org.apache.nifi.processor.util.list.ListedEntityTracker.TRACKING_STATE_CACHE; +import static org.apache.nifi.processor.util.list.ListedEntityTracker.TRACKING_TIME_WINDOW; +import static org.apache.nifi.processors.azure.AbstractAzureDataLakeStorageProcessor.ADLS_CREDENTIALS_SERVICE; +import static org.apache.nifi.processors.azure.AbstractAzureDataLakeStorageProcessor.DIRECTORY; +import static org.apache.nifi.processors.azure.AbstractAzureDataLakeStorageProcessor.FILESYSTEM; +import static org.apache.nifi.processors.azure.AbstractAzureDataLakeStorageProcessor.getStorageClient; + +@PrimaryNodeOnly +@TriggerSerially +@Tags({"azure", "microsoft", "cloud", "storage", "adlsgen2", "datalake"}) +@SeeAlso({PutAzureDataLakeStorage.class, DeleteAzureDataLakeStorage.class, FetchAzureDataLakeStorage.class}) +@CapabilityDescription("Lists directory in an Azure Data Lake Storage Gen 2 filesystem") +@WritesAttributes({ + @WritesAttribute(attribute = "azure.filesystem", description = "The name of the Azure File System"), + @WritesAttribute(attribute = "azure.filePath", description = "The full path of the Azure File"), + @WritesAttribute(attribute = "azure.directory", description = "The name of the Azure Directory"), + @WritesAttribute(attribute = "azure.filename", description = "The name of the Azure File"), + @WritesAttribute(attribute = "azure.length", description = "The length of the Azure File"), + @WritesAttribute(attribute = "azure.lastModified", description = "The last modification time of the Azure File"), + @WritesAttribute(attribute = "azure.etag", description = "The ETag of the Azure File") +}) +@InputRequirement(Requirement.INPUT_FORBIDDEN) +@Stateful(scopes = {Scope.CLUSTER}, description = "After performing a listing of files, the timestamp of the newest file is stored. " + + "This allows the Processor to list only files that have been added or modified after this date the next time that the Processor is run. State is " + + "stored across the cluster so that this Processor can be run on Primary Node only and if a new Primary Node is selected, the new node can pick up " + + "where the previous node left off, without duplicating the data.") +public class ListAzureDataLakeStorage extends AbstractListProcessor<ADLSFileInfo> { + + public static final PropertyDescriptor RECURSE_SUBDIRECTORIES = new PropertyDescriptor.Builder() + .name("recurse-subdirectories") + .displayName("Recurse Subdirectories") + .description("Indicates whether to list files from subdirectories of the directory") + .required(true) + .allowableValues("true", "false") + .defaultValue("true") + .build(); + + public static final PropertyDescriptor FILE_FILTER = new PropertyDescriptor.Builder() + .name("file-filter") + .displayName("File Filter") + .description("Only files whose names match the given regular expression will be picked up") Review comment: done ---------------------------------------------------------------- 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
