Repository: nifi Updated Branches: refs/heads/master e8f6ff440 -> 62e388aa4
NIFI-4709 - Fixed ListAzureBlobStorage timestamp precision handling. Signed-off-by: Pierre Villard <[email protected]> This closes #2354. Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/62e388aa Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/62e388aa Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/62e388aa Branch: refs/heads/master Commit: 62e388aa4fd0216abe7626902612196e7c9e41c8 Parents: e8f6ff4 Author: Koji Kawamura <[email protected]> Authored: Tue Dec 19 18:25:23 2017 +0900 Committer: Pierre Villard <[email protected]> Committed: Thu Dec 21 15:15:09 2017 +0100 ---------------------------------------------------------------------- .../azure/storage/ListAzureBlobStorage.java | 8 +++++++ .../util/list/AbstractListProcessor.java | 22 +++++++++++++++++++- 2 files changed, 29 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/62e388aa/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java index 7024e22..41f347c 100644 --- a/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java +++ b/nifi-nar-bundles/nifi-azure-bundle/nifi-azure-processors/src/main/java/org/apache/nifi/processors/azure/storage/ListAzureBlobStorage.java @@ -135,6 +135,14 @@ public class ListAzureBlobStorage extends AbstractListProcessor<BlobInfo> { } @Override + protected String getDefaultTimePrecision() { + // User does not have to choose one. + // AUTO_DETECT can handle most cases, but it may incur longer latency + // when all listed files do not have SECOND part in their timestamps although Azure Blob Storage does support seconds. + return PRECISION_SECONDS.getValue(); + } + + @Override protected List<BlobInfo> performListing(final ProcessContext context, final Long minTimestamp) throws IOException { String containerName = context.getProperty(AzureStorageUtils.CONTAINER).evaluateAttributeExpressions().getValue(); String prefix = context.getProperty(PROP_PREFIX).evaluateAttributeExpressions().getValue(); http://git-wip-us.apache.org/repos/asf/nifi/blob/62e388aa/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/list/AbstractListProcessor.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/list/AbstractListProcessor.java b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/list/AbstractListProcessor.java index 76e660d..31ba380 100644 --- a/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/list/AbstractListProcessor.java +++ b/nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/list/AbstractListProcessor.java @@ -41,6 +41,7 @@ import org.apache.nifi.processor.ProcessContext; import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.util.StringUtils; import java.io.File; import java.io.FileInputStream; @@ -124,6 +125,10 @@ import java.util.stream.Collectors; * changing a property value. The {@link #isListingResetNecessary(PropertyDescriptor)} method is responsible for determining when the listing needs to be reset by returning * a boolean indicating whether or not a change in the value of the provided property should trigger the timestamp and identifier information to be cleared. * </li> + * <li> + * Provide the target system timestamp precision. By either letting user to choose the right one by adding TARGET_SYSTEM_TIMESTAMP_PRECISION to the return value of + * getSupportedPropertyDescriptors method or, overriding getDefaultTimePrecision method in case the target system has a fixed time precision. + * </li> * </ul> */ @TriggerSerially @@ -438,7 +443,11 @@ public abstract class AbstractListProcessor<T extends ListableEntity> extends Ab latestListedEntryTimestampThisCycleMillis = orderedEntries.lastKey(); // Determine target system time precision. - final String specifiedPrecision = context.getProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION).getValue(); + String specifiedPrecision = context.getProperty(TARGET_SYSTEM_TIMESTAMP_PRECISION).getValue(); + if (StringUtils.isBlank(specifiedPrecision)) { + // If TARGET_SYSTEM_TIMESTAMP_PRECISION is not supported by the Processor, then specifiedPrecision can be null, instead of its default value. + specifiedPrecision = getDefaultTimePrecision(); + } final TimeUnit targetSystemTimePrecision = PRECISION_AUTO_DETECT.getValue().equals(specifiedPrecision) ? targetSystemHasMilliseconds ? TimeUnit.MILLISECONDS : targetSystemHasSeconds ? TimeUnit.SECONDS : TimeUnit.MINUTES @@ -544,6 +553,17 @@ public abstract class AbstractListProcessor<T extends ListableEntity> extends Ab } } + /** + * This method is intended to be overridden by SubClasses those do not support TARGET_SYSTEM_TIMESTAMP_PRECISION property. + * So that it use return different precisions than PRECISION_AUTO_DETECT. + * If TARGET_SYSTEM_TIMESTAMP_PRECISION is supported as a valid Processor property, + * then PRECISION_AUTO_DETECT will be the default value when not specified by a user. + * @return + */ + protected String getDefaultTimePrecision() { + return TARGET_SYSTEM_TIMESTAMP_PRECISION.getDefaultValue(); + } + private void resetTimeStates() { lastListedLatestEntryTimestampMillis = null; lastProcessedLatestEntryTimestampMillis = 0L;
