You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "exceptionfactory (via GitHub)" <gi...@apache.org> on 2023/12/22 19:40:06 UTC
Re: [PR] NIFI-12441 Added No Tracking listing strategy to ListS3 [nifi]
exceptionfactory commented on code in PR #8088:
URL: https://github.com/apache/nifi/pull/8088#discussion_r1435309260
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/ListS3.java:
##########
@@ -451,11 +456,92 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
listByTrackingTimestamps(context, session);
} else if (BY_ENTITIES.equals(listingStrategy)) {
listByTrackingEntities(context, session);
+ } else if (NO_TRACKING.equals(listingStrategy)) {
+ listNoTracking(context, session);
} else {
throw new ProcessException("Unknown listing strategy: " + listingStrategy);
}
}
+ private void listNoTracking(ProcessContext context, ProcessSession session) {
+ final AmazonS3 client = getClient(context);
+
+ S3BucketLister bucketLister = getS3BucketLister(context, client);
+
+ final long startNanos = System.nanoTime();
+ final long minAgeMilliseconds = context.getProperty(MIN_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
+ final Long maxAgeMilliseconds = context.getProperty(MAX_AGE) != null ? context.getProperty(MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS) : null;
+ final long listingTimestamp = System.currentTimeMillis();
+
+ final String bucket = context.getProperty(BUCKET_WITHOUT_DEFAULT_VALUE).evaluateAttributeExpressions().getValue();
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+
+ int listCount = 0;
+ int totalListCount = 0;
+
+ getLogger().trace("Start listing, listingTimestamp={}", new Object[]{listingTimestamp});
+
+ final S3ObjectWriter writer;
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ if (writerFactory == null) {
+ writer = new AttributeObjectWriter(session);
+ } else {
+ writer = new RecordObjectWriter(session, writerFactory, getLogger(), context.getProperty(S3_REGION).getValue());
+ }
+
+ try {
+ writer.beginListing();
+
+ do {
+ VersionListing versionListing = bucketLister.listVersions();
+ for (S3VersionSummary versionSummary : versionListing.getVersionSummaries()) {
+ long lastModified = versionSummary.getLastModified().getTime();
+ if ((maxAgeMilliseconds != null && (lastModified < (listingTimestamp - maxAgeMilliseconds)))
+ || lastModified > (listingTimestamp - minAgeMilliseconds)) {
+ continue;
+ }
+
+ getLogger().trace("Listed key={}, lastModified={}", new Object[]{versionSummary.getKey(), lastModified});
Review Comment:
```suggestion
getLogger().trace("Listed key={}, lastModified={}", versionSummary.getKey(), lastModified);
```
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/ListS3.java:
##########
@@ -451,11 +456,92 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
listByTrackingTimestamps(context, session);
} else if (BY_ENTITIES.equals(listingStrategy)) {
listByTrackingEntities(context, session);
+ } else if (NO_TRACKING.equals(listingStrategy)) {
+ listNoTracking(context, session);
} else {
throw new ProcessException("Unknown listing strategy: " + listingStrategy);
}
}
+ private void listNoTracking(ProcessContext context, ProcessSession session) {
+ final AmazonS3 client = getClient(context);
+
+ S3BucketLister bucketLister = getS3BucketLister(context, client);
+
+ final long startNanos = System.nanoTime();
+ final long minAgeMilliseconds = context.getProperty(MIN_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
+ final Long maxAgeMilliseconds = context.getProperty(MAX_AGE) != null ? context.getProperty(MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS) : null;
+ final long listingTimestamp = System.currentTimeMillis();
+
+ final String bucket = context.getProperty(BUCKET_WITHOUT_DEFAULT_VALUE).evaluateAttributeExpressions().getValue();
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+
+ int listCount = 0;
+ int totalListCount = 0;
+
+ getLogger().trace("Start listing, listingTimestamp={}", new Object[]{listingTimestamp});
Review Comment:
The `Object[]` wrapper is not necessary for log arguments, although not all components have been updated to reflect the recommended usage, new code should follow the pattern.
```suggestion
getLogger().trace("Start listing, listingTimestamp={}", listingTimestamp);
```
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/ListS3.java:
##########
@@ -451,11 +456,92 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
listByTrackingTimestamps(context, session);
} else if (BY_ENTITIES.equals(listingStrategy)) {
listByTrackingEntities(context, session);
+ } else if (NO_TRACKING.equals(listingStrategy)) {
+ listNoTracking(context, session);
} else {
throw new ProcessException("Unknown listing strategy: " + listingStrategy);
}
}
+ private void listNoTracking(ProcessContext context, ProcessSession session) {
+ final AmazonS3 client = getClient(context);
+
+ S3BucketLister bucketLister = getS3BucketLister(context, client);
+
+ final long startNanos = System.nanoTime();
+ final long minAgeMilliseconds = context.getProperty(MIN_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
+ final Long maxAgeMilliseconds = context.getProperty(MAX_AGE) != null ? context.getProperty(MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS) : null;
+ final long listingTimestamp = System.currentTimeMillis();
+
+ final String bucket = context.getProperty(BUCKET_WITHOUT_DEFAULT_VALUE).evaluateAttributeExpressions().getValue();
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+
+ int listCount = 0;
+ int totalListCount = 0;
+
+ getLogger().trace("Start listing, listingTimestamp={}", new Object[]{listingTimestamp});
+
+ final S3ObjectWriter writer;
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ if (writerFactory == null) {
+ writer = new AttributeObjectWriter(session);
+ } else {
+ writer = new RecordObjectWriter(session, writerFactory, getLogger(), context.getProperty(S3_REGION).getValue());
+ }
+
+ try {
+ writer.beginListing();
+
+ do {
+ VersionListing versionListing = bucketLister.listVersions();
+ for (S3VersionSummary versionSummary : versionListing.getVersionSummaries()) {
+ long lastModified = versionSummary.getLastModified().getTime();
+ if ((maxAgeMilliseconds != null && (lastModified < (listingTimestamp - maxAgeMilliseconds)))
+ || lastModified > (listingTimestamp - minAgeMilliseconds)) {
+ continue;
+ }
+
+ getLogger().trace("Listed key={}, lastModified={}", new Object[]{versionSummary.getKey(), lastModified});
+
+ GetObjectTaggingResult taggingResult = getTaggingResult(context, client, versionSummary);
+
+ ObjectMetadata objectMetadata = getObjectMetadata(context, client, versionSummary);
+
+ // Write the entity to the listing
+ writer.addToListing(versionSummary, taggingResult, objectMetadata, context.getProperty(S3_REGION).getValue());
+
+ listCount++;
+ }
+ bucketLister.setNextMarker();
+
+ totalListCount += listCount;
+
+ if (listCount >= batchSize && writer.isCheckpoint()) {
+ getLogger().info("Successfully listed {} new files from S3; routing to success", listCount);
+ session.commitAsync();
+ }
+
+ listCount = 0;
+ } while (bucketLister.isTruncated());
+
+ writer.finishListing();
+ } catch (final Exception e) {
+ getLogger().error("Failed to list contents of bucket due to {}", e, e);
+ writer.finishListingExceptionally(e);
+ session.rollback();
+ context.yield();
+ return;
+ }
+
+ final long listMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ getLogger().info("Successfully listed S3 bucket {} in {} millis", new Object[]{bucket, listMillis});
Review Comment:
```suggestion
getLogger().info("Successfully listed S3 bucket {} in {} millis", bucket, listMillis);
```
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/ListS3.java:
##########
@@ -451,11 +456,92 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
listByTrackingTimestamps(context, session);
} else if (BY_ENTITIES.equals(listingStrategy)) {
listByTrackingEntities(context, session);
+ } else if (NO_TRACKING.equals(listingStrategy)) {
+ listNoTracking(context, session);
} else {
throw new ProcessException("Unknown listing strategy: " + listingStrategy);
}
}
+ private void listNoTracking(ProcessContext context, ProcessSession session) {
+ final AmazonS3 client = getClient(context);
+
+ S3BucketLister bucketLister = getS3BucketLister(context, client);
+
+ final long startNanos = System.nanoTime();
+ final long minAgeMilliseconds = context.getProperty(MIN_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
+ final Long maxAgeMilliseconds = context.getProperty(MAX_AGE) != null ? context.getProperty(MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS) : null;
+ final long listingTimestamp = System.currentTimeMillis();
+
+ final String bucket = context.getProperty(BUCKET_WITHOUT_DEFAULT_VALUE).evaluateAttributeExpressions().getValue();
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+
+ int listCount = 0;
+ int totalListCount = 0;
+
+ getLogger().trace("Start listing, listingTimestamp={}", new Object[]{listingTimestamp});
+
+ final S3ObjectWriter writer;
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ if (writerFactory == null) {
+ writer = new AttributeObjectWriter(session);
+ } else {
+ writer = new RecordObjectWriter(session, writerFactory, getLogger(), context.getProperty(S3_REGION).getValue());
+ }
+
+ try {
+ writer.beginListing();
+
+ do {
+ VersionListing versionListing = bucketLister.listVersions();
+ for (S3VersionSummary versionSummary : versionListing.getVersionSummaries()) {
+ long lastModified = versionSummary.getLastModified().getTime();
+ if ((maxAgeMilliseconds != null && (lastModified < (listingTimestamp - maxAgeMilliseconds)))
+ || lastModified > (listingTimestamp - minAgeMilliseconds)) {
+ continue;
+ }
+
+ getLogger().trace("Listed key={}, lastModified={}", new Object[]{versionSummary.getKey(), lastModified});
+
+ GetObjectTaggingResult taggingResult = getTaggingResult(context, client, versionSummary);
+
+ ObjectMetadata objectMetadata = getObjectMetadata(context, client, versionSummary);
+
+ // Write the entity to the listing
+ writer.addToListing(versionSummary, taggingResult, objectMetadata, context.getProperty(S3_REGION).getValue());
+
+ listCount++;
+ }
+ bucketLister.setNextMarker();
+
+ totalListCount += listCount;
+
+ if (listCount >= batchSize && writer.isCheckpoint()) {
+ getLogger().info("Successfully listed {} new files from S3; routing to success", listCount);
+ session.commitAsync();
+ }
+
+ listCount = 0;
+ } while (bucketLister.isTruncated());
+
+ writer.finishListing();
+ } catch (final Exception e) {
+ getLogger().error("Failed to list contents of bucket due to {}", e, e);
+ writer.finishListingExceptionally(e);
+ session.rollback();
+ context.yield();
+ return;
+ }
+
+ final long listMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+ getLogger().info("Successfully listed S3 bucket {} in {} millis", new Object[]{bucket, listMillis});
+
+ if (totalListCount == 0) {
+ getLogger().debug("No new objects in S3 bucket {} to list. Yielding.", new Object[]{bucket});
Review Comment:
```suggestion
getLogger().debug("No new objects in S3 bucket {} to list. Yielding", bucket);
```
##########
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/s3/ListS3.java:
##########
@@ -451,11 +456,92 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
listByTrackingTimestamps(context, session);
} else if (BY_ENTITIES.equals(listingStrategy)) {
listByTrackingEntities(context, session);
+ } else if (NO_TRACKING.equals(listingStrategy)) {
+ listNoTracking(context, session);
} else {
throw new ProcessException("Unknown listing strategy: " + listingStrategy);
}
}
+ private void listNoTracking(ProcessContext context, ProcessSession session) {
+ final AmazonS3 client = getClient(context);
+
+ S3BucketLister bucketLister = getS3BucketLister(context, client);
+
+ final long startNanos = System.nanoTime();
+ final long minAgeMilliseconds = context.getProperty(MIN_AGE).asTimePeriod(TimeUnit.MILLISECONDS);
+ final Long maxAgeMilliseconds = context.getProperty(MAX_AGE) != null ? context.getProperty(MAX_AGE).asTimePeriod(TimeUnit.MILLISECONDS) : null;
+ final long listingTimestamp = System.currentTimeMillis();
+
+ final String bucket = context.getProperty(BUCKET_WITHOUT_DEFAULT_VALUE).evaluateAttributeExpressions().getValue();
+ final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
+
+ int listCount = 0;
+ int totalListCount = 0;
+
+ getLogger().trace("Start listing, listingTimestamp={}", new Object[]{listingTimestamp});
+
+ final S3ObjectWriter writer;
+ final RecordSetWriterFactory writerFactory = context.getProperty(RECORD_WRITER).asControllerService(RecordSetWriterFactory.class);
+ if (writerFactory == null) {
+ writer = new AttributeObjectWriter(session);
+ } else {
+ writer = new RecordObjectWriter(session, writerFactory, getLogger(), context.getProperty(S3_REGION).getValue());
+ }
+
+ try {
+ writer.beginListing();
+
+ do {
+ VersionListing versionListing = bucketLister.listVersions();
+ for (S3VersionSummary versionSummary : versionListing.getVersionSummaries()) {
+ long lastModified = versionSummary.getLastModified().getTime();
+ if ((maxAgeMilliseconds != null && (lastModified < (listingTimestamp - maxAgeMilliseconds)))
+ || lastModified > (listingTimestamp - minAgeMilliseconds)) {
+ continue;
+ }
+
+ getLogger().trace("Listed key={}, lastModified={}", new Object[]{versionSummary.getKey(), lastModified});
+
+ GetObjectTaggingResult taggingResult = getTaggingResult(context, client, versionSummary);
+
+ ObjectMetadata objectMetadata = getObjectMetadata(context, client, versionSummary);
+
+ // Write the entity to the listing
+ writer.addToListing(versionSummary, taggingResult, objectMetadata, context.getProperty(S3_REGION).getValue());
+
+ listCount++;
+ }
+ bucketLister.setNextMarker();
+
+ totalListCount += listCount;
+
+ if (listCount >= batchSize && writer.isCheckpoint()) {
+ getLogger().info("Successfully listed {} new files from S3; routing to success", listCount);
+ session.commitAsync();
+ }
+
+ listCount = 0;
+ } while (bucketLister.isTruncated());
+
+ writer.finishListing();
+ } catch (final Exception e) {
+ getLogger().error("Failed to list contents of bucket due to {}", e, e);
Review Comment:
The `due to {}` approach is duplicative and also needs to be removed from older implementations, but this is an opportunity to streamline the log and avoid duplicating the message, as it is already included in the stack trace.
```suggestion
getLogger().error("Failed to list contents of bucket", e);
```
--
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.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org