NIFI-14543 Add record counter in ConsumeBoxEnterpriseEvents

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

This closes #9920.

Author: awelless

nifi-extension-bundles/nifi-box-bundle/nifi-box-processors/src/main/java/org/apache/nifi/processors/box/ConsumeBoxEnterpriseEvents.java

@@ -74,6 +74,8 @@ public class ConsumeBoxEnterpriseEvents extends AbstractProcessor {
 
     private static final int LIMIT = 500;
 
+    static final String COUNTER_RECORDS_PROCESSED = "Records Processed";
+
     static final PropertyDescriptor EVENT_TYPES = new PropertyDescriptor.Builder()
             .name("Event Types")
             .description("A comma separated list of Enterprise Events to consume. If not set, all Events are consumed." +
@@ -220,6 +222,7 @@ private void writeLogAsRecords(final EventLog eventLog, final ProcessSession ses
             throw new ProcessException("Failed to write Box Event into a FlowFile", e);
         }
 
+        session.adjustCounter(COUNTER_RECORDS_PROCESSED, eventLog.getSize(), false);
         session.putAttribute(flowFile, "record.count", String.valueOf(eventLog.getSize()));
         session.putAttribute(flowFile, CoreAttributes.MIME_TYPE.key(), "application/json");
         session.transfer(flowFile, REL_SUCCESS);

nifi-extension-bundles/nifi-box-bundle/nifi-box-processors/src/test/java/org/apache/nifi/processors/box/ConsumeBoxEnterpriseEventsTest.java

@@ -92,6 +92,8 @@ void testConsumeEvents(
                 .toList();
 
         assertEquals(expectedEventIds, eventIds);
+
+        assertEquals(eventIds.size(), testRunner.getCounterValue(ConsumeBoxEnterpriseEvents.COUNTER_RECORDS_PROCESSED));
     }
 
     static List<Arguments> dataFor_testConsumeEvents() {
@@ -129,6 +131,7 @@ void testGracefulTermination() throws InterruptedException {
             assertDoesNotThrow(() -> runFuture.get(5, TimeUnit.SECONDS), "Processor did not stop gracefully");
 
             testRunner.assertAllFlowFilesTransferred(ConsumeBoxEnterpriseEvents.REL_SUCCESS, consumedEvents.get());
+            assertEquals(consumedEvents.get(), testRunner.getCounterValue(ConsumeBoxEnterpriseEvents.COUNTER_RECORDS_PROCESSED));
         } finally {
             // We can't use try with resources, as Executors use a shutdown method
             // which indefinitely waits for submitted tasks.