turcsanyip commented on a change in pull request #4438:
URL: https://github.com/apache/nifi/pull/4438#discussion_r463880093
##########
File path:
nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/AbstractAzureDataLakeStorageProcessor.java
##########
@@ -146,8 +154,22 @@ public static DataLakeServiceClient
getStorageClient(PropertyContext context, Fl
return storageClient;
}
- @Override
- public Set<Relationship> getRelationships() {
- return RELATIONSHIPS;
+ private static class DirectoryValidator implements Validator {
+ @Override
+ public ValidationResult validate(String subject, String input,
ValidationContext context) {
+ ValidationResult.Builder builder = new ValidationResult.Builder()
+ .subject(DIRECTORY.getDisplayName())
+ .input(input);
+
+ if (context.isExpressionLanguagePresent(input)) {
+ builder.valid(true).explanation("Expression Language Present");
+ } else if (input.startsWith("/")) {
+ builder.valid(false).explanation(String.format("'%s' cannot
contain a leading '/'", DIRECTORY.getDisplayName()));
+ } else {
+ builder.valid(true);
Review comment:
Added whitespace-only check.
##########
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")
+ .required(false)
+ .addValidator(StandardValidators.REGULAR_EXPRESSION_VALIDATOR)
+ .build();
+
+ public static final PropertyDescriptor PATH_FILTER = new
PropertyDescriptor.Builder()
+ .name("path-filter")
+ .displayName("Path Filter")
+ .description("When " + RECURSE_SUBDIRECTORIES.getName() + " is
true, then only subdirectories whose paths match the given regular expression
will be scanned")
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:
[email protected]
