[GitHub] [nifi] mattyb149 commented on a change in pull request #5220: NIFI-8790 allow Expression Language for Index Operation in PutElasticsearchRecord

[GitBox <gi...@apache.org>]

mattyb149 commented on a change in pull request #5220:
URL: https://github.com/apache/nifi/pull/5220#discussion_r674339249



##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -183,6 +179,33 @@ public void onScheduled(ProcessContext context) {
         this.logErrors = context.getProperty(LOG_ERROR_RESPONSES).asBoolean();
     }
 
+    static final List<String> ALLOWED_INDEX_OPERATIONS = Collections.unmodifiableList(Arrays.asList(
+            IndexOperationRequest.Operation.Create.getValue(),
+            IndexOperationRequest.Operation.Delete.getValue(),
+            IndexOperationRequest.Operation.Index.getValue(),
+            IndexOperationRequest.Operation.Update.getValue(),
+            IndexOperationRequest.Operation.Upsert.getValue()
+    ));
+
+    @Override
+    protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+        final List<ValidationResult> validationResults = new ArrayList<>();
+
+        final PropertyValue indexOp = validationContext.getProperty(INDEX_OP);
+        if (indexOp.isSet() && !indexOp.isExpressionLanguagePresent()) {

Review comment:
       IIRC if EL is present we still return a ValidationResult, but it's marked VALID and has text like "Expression Language present". It's a little more clear that way and might help if their EL statement evaluates to something invalid, the user will wonder why the processor was valid but can see that we didn't evaluate the EL because we'll display the "valid" ValidationResult.

##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -88,16 +92,8 @@
             .name("put-es-record-index-op")
             .displayName("Index Operation")
             .description("The type of the operation used to index (create, delete, index, update, upsert)")
+            .addValidator(StandardValidators.NON_EMPTY_EL_VALIDATOR)
             .expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
-            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
-            .required(false)

Review comment:
       This should probably be `required(true)` now, I believe it was overlooked before because with Allowed Values you always have to pick one. Having a NON_EMPTY_EL_VALIDATOR will work the same, but it might help to have the property in bold font as required anyway.
   
   Another thing to consider is that other processors that allow EL but also preserve the drop-down list will instead add a choice of something like "Use 'es.operation.type' Attribute" so the user can still control the operation dynamically but don't need an EL statement in the processor itself. It doesn't really matter to me which way you go but wanted to point it out.

##########
File path: nifi-nar-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java
##########
@@ -297,17 +321,20 @@ private FlowFile indexDocuments(BulkOperation bundle, ProcessSession session, Fl
             if (writerFactory != null) {
                 FlowFile errorFF = session.create(input);
                 try (OutputStream os = session.write(errorFF);
-                     RecordSetWriter writer = writerFactory.createWriter(getLogger(), bundle.getSchema(), os )) {
+                     RecordSetWriter writer = writerFactory.createWriter(getLogger(), bundle.getSchema(), os, errorFF )) {
 
                     int added = 0;
                     writer.beginRecordSet();
                     for (int index = 0; index < response.getItems().size(); index++) {
                         Map<String, Object> current = response.getItems().get(index);
-                        String key = current.keySet().stream().findFirst().get();
-                        Map<String, Object> inner = (Map<String, Object>) current.get(key);
-                        if (inner.containsKey("error")) {
-                            writer.write(bundle.getOriginalRecords().get(index));
-                            added++;
+                        if (!current.isEmpty()) {
+                            String key = current.keySet().stream().findFirst().get();

Review comment:
       I realize this not really a changed line but I believe it would be faster to use iterator().next() instead of a stream (the underlying implementation is unfortunately slow)




-- 
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