From 735752a28587535c3877056df992dc0efc095269 Mon Sep 17 00:00:00 2001 From: Alessandro D'Armiento Date: Tue, 23 Jul 2019 15:00:11 +0200 Subject: [PATCH] [#6465] ListHDFS: skipping last instant files should be optional

Changes

* ListHDFS accepts a new optional property "Skip Last" * Default value is true, which maintain the existing behaviour untouched * If Skip Last is set to false, avoid the consistency logic and always return the full file listing --- .../nifi/processors/hadoop/ListHDFS.java | 58 +++++++++++++------ .../nifi/processors/hadoop/TestListHDFS.java | 17 ++++++ 2 files changed, 57 insertions(+), 18 deletions(-) diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java index 27f58ef6b696..aee74b8b0a94 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/main/java/org/apache/nifi/processors/hadoop/ListHDFS.java @@ -40,6 +40,7 @@ import org.apache.nifi.components.state.Scope; import org.apache.nifi.components.state.StateMap; import org.apache.nifi.distributed.cache.client.DistributedMapCacheClient; +import org.apache.nifi.expression.ExpressionLanguageScope; import org.apache.nifi.flowfile.FlowFile; import org.apache.nifi.flowfile.attributes.CoreAttributes; import org.apache.nifi.processor.ProcessContext; @@ -69,8 +70,9 @@ @TriggerWhenEmpty @InputRequirement(Requirement.INPUT_FORBIDDEN) @Tags({"hadoop", "HDFS", "get", "list", "ingest", "source", "filesystem"}) -@CapabilityDescription("Retrieves a listing of files from HDFS. Each time a listing is performed, the files with the latest timestamp will be excluded " - + "and picked up during the next execution of the processor. This is done to ensure that we do not miss any files, or produce duplicates, in the " +@CapabilityDescription("Retrieves a listing of files from HDFS. Each time a listing is performed, if \"Skip Last\" is enabled, " + + "the files with the latest timestamp will be excluded and picked up during the next execution of the processor. " + + "This is done to ensure that we do not miss any files, or produce duplicates, in the " + "cases where files with the same timestamp are written immediately before and after a single execution of the processor. For each file that is " + "listed in HDFS, this processor creates a FlowFile that represents the HDFS file to be fetched in conjunction with FetchHDFS. This Processor is " + "designed to run on Primary Node only in a cluster. If the primary node changes, the new Primary Node will pick up where the previous node left " @@ -167,6 +169,20 @@ public class ListHDFS extends AbstractHadoopProcessor { .addValidator(StandardValidators.createTimePeriodValidator(100, TimeUnit.MILLISECONDS, Long.MAX_VALUE, TimeUnit.NANOSECONDS)) .build(); + public static final PropertyDescriptor SKIP_LAST = new PropertyDescriptor.Builder() + .name("Skip Last") + .description("If enabled, the files with the latest timestamp will be excluded and picked up during the next " + + "execution of the processor. This is done to ensure that we do not miss any files, or produce duplicates, " + + "in the cases where files with the same timestamp are written immediately before and after a single execution of the processor." + + "\nNOTE: this flag should be turned off only in case the listing directory content is not being modified, for example after " + + "the completion of a job which populated that directory, otherwise, consistency cannot be guaranteed.") + .required(true) + .allowableValues("true", "false") + .defaultValue("true") + .expressionLanguageSupported(ExpressionLanguageScope.VARIABLE_REGISTRY) + .addValidator(StandardValidators.NON_EMPTY_VALIDATOR) + .build(); + public static final Relationship REL_SUCCESS = new Relationship.Builder() .name("success") .description("All FlowFiles are transferred to this relationship") @@ -200,6 +216,7 @@ protected List getSupportedPropertyDescriptors() { props.add(FILE_FILTER_MODE); props.add(MIN_AGE); props.add(MAX_AGE); + props.add(SKIP_LAST); return props; } @@ -246,7 +263,7 @@ public void onPropertyModified(final PropertyDescriptor descriptor, final String * @param context processor context with properties values * @return a Set containing only those FileStatus objects that we want to list */ - Set determineListable(final Set statuses, ProcessContext context) { + Set determineListable(final Set statuses, ProcessContext context, boolean skipLast) { final long minTimestamp = this.latestTimestampListed; final TreeMap> orderedEntries = new TreeMap<>(); @@ -292,26 +309,30 @@ Set determineListable(final Set statuses, ProcessContext if (orderedEntries.size() > 0) { long latestListingTimestamp = orderedEntries.lastKey(); - // If the last listing time is equal to the newest entries previously seen, - // another iteration has occurred without new files and special handling is needed to avoid starvation - if (latestListingTimestamp == minTimestamp) { - // We are done if the latest listing timestamp is equal to the last processed time, - // meaning we handled those items originally passed over - if (latestListingTimestamp == latestTimestampEmitted) { - return Collections.emptySet(); + if (skipLast) { + // If the last listing time is equal to the newest entries previously seen, + // another iteration has occurred without new files and special handling is needed to avoid starvation + if (latestListingTimestamp == minTimestamp) { + // We are done if the latest listing timestamp is equal to the last processed time, + // meaning we handled those items originally passed over + if (latestListingTimestamp == latestTimestampEmitted) { + return Collections.emptySet(); + } + } else { + // Otherwise, newest entries are held back one cycle to avoid issues in writes occurring exactly when the listing is being performed to avoid missing data + orderedEntries.remove(latestListingTimestamp); } - } else { - // Otherwise, newest entries are held back one cycle to avoid issues in writes occurring exactly when the listing is being performed to avoid missing data - orderedEntries.remove(latestListingTimestamp); - } - for (List timestampEntities : orderedEntries.values()) { - for (FileStatus status : timestampEntities) { - toList.add(status); + } else { + if (latestListingTimestamp == minTimestamp && latestListingTimestamp == latestTimestampEmitted) { + return Collections.emptySet(); } } } + for (List timestampEntities : orderedEntries.values()) { + toList.addAll(timestampEntities); + } return toList; } @@ -329,6 +350,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session lastRunTimestamp = now; final String directory = context.getProperty(DIRECTORY).evaluateAttributeExpressions().getValue(); + final Boolean skipLast = context.getProperty(SKIP_LAST).evaluateAttributeExpressions().asBoolean(); // Ensure that we are using the latest listing information before we try to perform a listing of HDFS files. try { @@ -387,7 +409,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session return; } - final Set listable = determineListable(statuses, context); + final Set listable = determineListable(statuses, context, skipLast); getLogger().debug("Of the {} files found in HDFS, {} are listable", new Object[] {statuses.size(), listable.size()}); for (final FileStatus status : listable) { diff --git a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java index 8134b4d870b3..c7a3a5fb74df 100644 --- a/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java +++ b/nifi-nar-bundles/nifi-hadoop-bundle/nifi-hdfs-processors/src/test/java/org/apache/nifi/processors/hadoop/TestListHDFS.java @@ -533,6 +533,23 @@ public void testListAfterDirectoryChange() throws InterruptedException { } + @Test + public void testListFilesDoesntSkipLastTimestampFilesWhenSkipLastIsFalse() { + proc.fileSystem.addFileStatus(new Path("/test"), new FileStatus(1L, false, 1, 1L, 100L,0L, create777(), "owner", "group", new Path("/test/testFile-1.txt"))); + proc.fileSystem.addFileStatus(new Path("/test"), new FileStatus(1L, false, 1, 1L, 150L,0L, create777(), "owner", "group", new Path("/test/testFile-2.txt"))); + proc.fileSystem.addFileStatus(new Path("/test"), new FileStatus(1L, false, 1, 1L, 200L,0L, create777(), "owner", "group", new Path("/test/testFile-3.txt"))); + + runner.setProperty(ListHDFS.DIRECTORY, "/test"); + runner.setProperty(ListHDFS.SKIP_LAST, "false"); + + runner.run(); // Latest file from /test should not be ignored + + // Assert output FlowFiles has been transferred to ListHDFS.REL_SUCCESS + final List output = runner.getFlowFilesForRelationship(ListHDFS.REL_SUCCESS); + assertEquals(output.size(), 3); + } + + private FsPermission create777() { return new FsPermission((short) 0777); }