exceptionfactory commented on code in PR #10136:
URL: https://github.com/apache/nifi/pull/10136#discussion_r2251478121
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java:
##########
@@ -701,46 +703,64 @@ private FlowFilePoll pollFlowFiles(final ProcessContext
context, final ProcessSe
final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
final FlowFileFilter dbcpServiceFlowFileFilter =
context.getProperty(CONNECTION_POOL).asControllerService(DBCPService.class).getFlowFileFilter(batchSize);
- List<FlowFile> flowFiles;
+ final List<FlowFile> validFlowFiles;
+ final List<FlowFile> selectedFlowFiles;
if (useTransactions) {
final TransactionalFlowFileFilter filter = new
TransactionalFlowFileFilter(dbcpServiceFlowFileFilter);
- flowFiles = session.get(filter);
+ selectedFlowFiles = session.get(filter);
fragmentedTransaction = filter.isFragmentedTransaction();
} else {
if (dbcpServiceFlowFileFilter == null) {
- flowFiles = session.get(batchSize);
+ selectedFlowFiles = session.get(batchSize);
} else {
- flowFiles = session.get(dbcpServiceFlowFileFilter);
+ selectedFlowFiles = session.get(dbcpServiceFlowFileFilter);
}
}
- if (flowFiles.isEmpty()) {
+ boolean selectedFlowFilesShouldHaveDataBaseNameButDont =
dbcpServiceFlowFileFilter != null
+ && !selectedFlowFiles.isEmpty()
+ &&
selectedFlowFiles.stream().findAny().get().getAttribute("database.name") ==
null;
Review Comment:
This complex conditional and name is a bit hard to follow on initial read,
although the concept makes sense. I think having a first-level condition based
on the presence of `dbcpServiceFlowFileFilter`, then checking for FlowFiles
missing `database.name` would be a bit easier to follow.
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java:
##########
@@ -701,46 +703,64 @@ private FlowFilePoll pollFlowFiles(final ProcessContext
context, final ProcessSe
final int batchSize = context.getProperty(BATCH_SIZE).asInteger();
final FlowFileFilter dbcpServiceFlowFileFilter =
context.getProperty(CONNECTION_POOL).asControllerService(DBCPService.class).getFlowFileFilter(batchSize);
- List<FlowFile> flowFiles;
+ final List<FlowFile> validFlowFiles;
+ final List<FlowFile> selectedFlowFiles;
if (useTransactions) {
final TransactionalFlowFileFilter filter = new
TransactionalFlowFileFilter(dbcpServiceFlowFileFilter);
- flowFiles = session.get(filter);
+ selectedFlowFiles = session.get(filter);
fragmentedTransaction = filter.isFragmentedTransaction();
} else {
if (dbcpServiceFlowFileFilter == null) {
- flowFiles = session.get(batchSize);
+ selectedFlowFiles = session.get(batchSize);
} else {
- flowFiles = session.get(dbcpServiceFlowFileFilter);
+ selectedFlowFiles = session.get(dbcpServiceFlowFileFilter);
}
}
- if (flowFiles.isEmpty()) {
+ boolean selectedFlowFilesShouldHaveDataBaseNameButDont =
dbcpServiceFlowFileFilter != null
+ && !selectedFlowFiles.isEmpty()
+ &&
selectedFlowFiles.stream().findAny().get().getAttribute("database.name") ==
null;
+
+ if (selectedFlowFilesShouldHaveDataBaseNameButDont) {
+ selectedFlowFiles.forEach(flowFile -> getLogger().warn(
+ "FlowFile {} is invalid because it's missing a
'database.name' attribute. Routing to '{}'.",
+ flowFile.getAttribute(CoreAttributes.UUID.key()),
+ REL_FAILURE.getName()
+ ));
Review Comment:
Recommend adjusting the message since the FlowFile UUID and other
information is included in the standard `toString()` method. I also recommend
removing the trailing `.` character.
```suggestion
selectedFlowFiles.forEach(flowFile -> getLogger().warn(
"{} missing attribute named [database.name] Routing to
[{}]",
flowFile,
REL_FAILURE.getName()
));
```
##########
nifi-extension-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutSQL.java:
##########
@@ -805,50 +825,50 @@ boolean isFragmentedTransactionReady(final List<FlowFile>
flowFiles, final Long
return true;
} else if (fragmentCount == null) {
throw illegal.apply("Cannot process %s because there are %d
FlowFiles with the same fragment.identifier "
- + "attribute but not all FlowFiles have a
fragment.count attribute", new Object[] {flowFile, flowFiles.size()});
+ + "attribute but not all FlowFiles have a
fragment.count attribute", new Object[]{flowFile, flowFiles.size()});
Review Comment:
This looks like a good opportunity to remove the unnecessary `new
Object[]{}` wrapper, which would be a helpful cleanup. If you prefer to avoid
that change, I would revert the spacing changes.
--
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]
