Fixed PutElasticsearchHttpRecord, same bug as JoltTransformRecord

apache · Dec 4, 2019 · d807aee · d807aee
1 parent 43f8f9d
commit d807aee
Showing 1 changed file with 9 additions and 11 deletions.
diff --git a/...rs/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java b/...rs/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchHttpRecord.java
@@ -576,17 +576,6 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
                         i++;
                     }
                 }
-
-                session.putAttribute(successFlowFile, "record.count", Integer.toString(recordCount - failures.size()));
-
-                // Normal behavior is to output with record.count. In order to not break backwards compatibility, set both here.
-                session.putAttribute(failedFlowFile, "record.count", Integer.toString(failures.size()));
-                session.putAttribute(failedFlowFile, "failure.count", Integer.toString(failures.size()));
-
-                session.transfer(successFlowFile, REL_SUCCESS);
-                session.transfer(failedFlowFile, REL_FAILURE);
-                session.remove(inputFlowFile);
-
             } catch (final IOException | SchemaNotFoundException | MalformedRecordException e) {
                 // We failed while handling individual failures. Not much else we can do other than log, and route the whole thing to failure.
                 getLogger().error("Failed to process {} during individual record failure handling; route whole FF to failure", new Object[] {flowFile, e});
@@ -597,7 +586,16 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
                 if (failedFlowFile != null) {
                     session.remove(failedFlowFile);
                 }
+                return;
             }
+            session.putAttribute(successFlowFile, "record.count", Integer.toString(recordCount - failures.size()));
+
+            // Normal behavior is to output with record.count. In order to not break backwards compatibility, set both here.
+            session.putAttribute(failedFlowFile, "record.count", Integer.toString(failures.size()));
+            session.putAttribute(failedFlowFile, "failure.count", Integer.toString(failures.size()));
+            session.transfer(successFlowFile, REL_SUCCESS);
+            session.transfer(failedFlowFile, REL_FAILURE);
+            session.remove(inputFlowFile);
         }
     }