gresockj commented on a change in pull request #5504:
URL: https://github.com/apache/nifi/pull/5504#discussion_r762614949
##########
File path:
nifi-nar-bundles/nifi-aws-bundle/nifi-aws-processors/src/main/java/org/apache/nifi/processors/aws/dynamodb/GetDynamoDB.java
##########
@@ -100,42 +104,75 @@
}
@Override
- public void onTrigger(final ProcessContext context, final ProcessSession
session) {
- List<FlowFile> flowFiles =
session.get(context.getProperty(BATCH_SIZE).evaluateAttributeExpressions().asInteger());
- if (flowFiles == null || flowFiles.size() == 0) {
- return;
- }
+ public List<ConfigVerificationResult> verify(final ProcessContext context,
final ComponentLog verificationLogger, final Map<String, String> attributes) {
+ final List<ConfigVerificationResult> results = new
ArrayList<>(super.verify(context, verificationLogger, attributes));
- Map<ItemKeys,FlowFile> keysToFlowFileMap = new HashMap<>();
+ final TableKeysAndAttributes tableKeysAndAttributes =
getTableKeysAndAttributes(context, attributes);
final String table =
context.getProperty(TABLE).evaluateAttributeExpressions().getValue();
- TableKeysAndAttributes tableKeysAndAttributes = new
TableKeysAndAttributes(table);
-
- final String hashKeyName =
context.getProperty(HASH_KEY_NAME).evaluateAttributeExpressions().getValue();
- final String rangeKeyName =
context.getProperty(RANGE_KEY_NAME).evaluateAttributeExpressions().getValue();
final String jsonDocument =
context.getProperty(JSON_DOCUMENT).evaluateAttributeExpressions().getValue();
- for (FlowFile flowFile : flowFiles) {
- final Object hashKeyValue = getValue(context, HASH_KEY_VALUE_TYPE,
HASH_KEY_VALUE, flowFile);
- final Object rangeKeyValue = getValue(context,
RANGE_KEY_VALUE_TYPE, RANGE_KEY_VALUE, flowFile);
+ if (tableKeysAndAttributes.getPrimaryKeys().isEmpty()) {
- if ( ! isHashKeyValueConsistent(hashKeyName, hashKeyValue,
session, flowFile)) {
- continue;
- }
+ results.add(new ConfigVerificationResult.Builder()
+ .outcome(Outcome.SKIPPED)
+ .verificationStepName("Get DynamoDB Items")
+ .explanation(String.format("Skipped getting DynamoDB items
because no primary keys would be included in retrieval"))
+ .build());
+ } else {
+ try {
+ final DynamoDB dynamoDB =
getDynamoDB(getConfiguration(context).getClient());
+ int totalCount = 0;
+ int jsonDocumentCount = 0;
- if ( ! isRangeKeyValueConsistent(rangeKeyName, rangeKeyValue,
session, flowFile) ) {
- continue;
- }
+ BatchGetItemOutcome result =
dynamoDB.batchGetItem(tableKeysAndAttributes);
Review comment:
Same comment as above with FetchS3Object -- unfortunately the DynamoDB
client doesn't provide such a check (the `getTable` method doesn't actually
make a call, only returns a new `Table` object). I'd be inclined to leave it
as is.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
