Fixes a false reporting bug for the invalidEventHandles counter (#6420)

Fixes a bug with the invalidEventHandles counter in the PipelineRunner. This metric was being counted for any event that is not a default event (ie. for aggregate events). This would happen even if there is no need to discard the event. This change should count this when aggregate events should be released but are not. We probably need some deeper investigation into how we can properly release aggregate events. But, for now this metric will be more accurate.

Also improves some code to reduce unnecessary variables, use final modifiers, and better legibility.

Signed-off-by: David Venable <dlv@amazon.com>

Author: dlvenable

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/pipeline/PipelineRunnerImpl.java

@@ -1,6 +1,10 @@
 /*
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
  */
 
 package org.opensearch.dataprepper.core.pipeline;
@@ -15,7 +19,6 @@
 import org.opensearch.dataprepper.model.event.DefaultEventHandle;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.event.EventHandle;
-import org.opensearch.dataprepper.model.event.InternalEventHandle;
 import org.opensearch.dataprepper.model.processor.Processor;
 import org.opensearch.dataprepper.model.record.Record;
 import org.slf4j.Logger;
@@ -33,8 +36,7 @@ public class PipelineRunnerImpl implements PipelineRunner {
     private static final Logger LOG = LoggerFactory.getLogger(PipelineRunnerImpl.class);
     private static final String INVALID_EVENT_HANDLES = "invalidEventHandles";
     private boolean isEmptyRecordsLogged = false;
-    @VisibleForTesting
-    final Counter invalidEventHandlesCounter;
+    private final Counter invalidEventHandlesCounter;
     private final Pipeline pipeline;
     private final PluginMetrics pluginMetrics;
     private final ProcessorProvider processorProvider;
@@ -75,18 +77,19 @@ Map.Entry<Collection, CheckpointState> readFromBuffer(Buffer buffer, Pipeline pi
     }
 
     @VisibleForTesting
-    void processAcknowledgements(List<Event> inputEvents, Collection<Record<Event>> outputRecords) {
-        Set<Event> outputEventsSet = outputRecords.stream().map(Record::getData).collect(Collectors.toSet());
+    void processAcknowledgements(final List<Event> inputEvents, final Collection<Record<Event>> outputRecords) {
+        final Set<Event> outputEventsSet = outputRecords.stream().map(Record::getData).collect(Collectors.toSet());
         // For each event in the input events list that is not present in the output events, send positive acknowledgement, if acknowledgements are enabled for it
         inputEvents.forEach(event -> {
-            EventHandle eventHandle = event.getEventHandle();
-            if (eventHandle != null && eventHandle instanceof DefaultEventHandle) {
-                InternalEventHandle internalEventHandle = (InternalEventHandle) eventHandle;
+            final EventHandle eventHandle = event.getEventHandle();
+            if (eventHandle != null) {
                 if (!outputEventsSet.contains(event)) {
-                    eventHandle.release(true);
+                    if (eventHandle instanceof DefaultEventHandle) {
+                        eventHandle.release(true);
+                    } else {
+                        invalidEventHandlesCounter.increment();
+                    }
                 }
-            } else if (eventHandle != null) {
-                invalidEventHandlesCounter.increment();
             }
         });
     }

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/pipeline/router/RouterFactory.java

@@ -1,17 +1,24 @@
 /*
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
  */
 
 package org.opensearch.dataprepper.core.pipeline.router;
 
 import org.opensearch.dataprepper.expression.ExpressionEvaluator;
 import org.opensearch.dataprepper.model.configuration.ConditionalRoute;
+import org.opensearch.dataprepper.model.event.Event;
 
 import java.util.Objects;
 import java.util.Set;
+import java.util.function.Consumer;
 
 public class RouterFactory {
+    private static final Consumer<Event> RELEASE_EVENT_ON_NO_ROUTE = event -> event.getEventHandle().release(true);
     private final ExpressionEvaluator expressionEvaluator;
     private final DataFlowComponentRouter dataFlowComponentRouter;
 
@@ -23,6 +30,6 @@ public class RouterFactory {
     public Router createRouter(final Set<ConditionalRoute> routes) {
         final RouteEventEvaluator routeEventEvaluator = new RouteEventEvaluator(expressionEvaluator, routes);
         return new Router(routeEventEvaluator, dataFlowComponentRouter,
-                event -> event.getEventHandle().release(true));
+                    RELEASE_EVENT_ON_NO_ROUTE);
     }
 }

data-prepper-core/src/test/java/org/opensearch/dataprepper/core/pipeline/PipelineRunnerTest.java

@@ -1,6 +1,10 @@
 /*
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
  */
 
 package org.opensearch.dataprepper.core.pipeline;
@@ -55,13 +59,18 @@
 import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
 import static org.mockito.Mockito.when;
 
 @ExtendWith(MockitoExtension.class)
 class PipelineRunnerTest {
     private static final int BUFFER_READ_TIMEOUT_MILLIS = 1000;
     private static final String MOCK_PIPELINE_NAME = "Test-Pipeline";
 
+    @Mock
+    private PluginMetrics pluginMetrics;
+    @Mock
+    private Counter counter;
     @Mock
     Pipeline pipeline;
     @Mock
@@ -91,12 +100,19 @@ private void setupPipeline(boolean shouldEnableAcknowledgements) {
     }
 
     private PipelineRunnerImpl createObjectUnderTest() {
-        return new PipelineRunnerImpl(pipeline, processorProvider);
+        try (final MockedStatic<PluginMetrics> pluginMetricsStatic = mockStatic(PluginMetrics.class)) {
+            pluginMetricsStatic.when(() -> PluginMetrics.fromNames("PipelineRunner", pipeline.getName()))
+                    .thenReturn(pluginMetrics);
+
+            return new PipelineRunnerImpl(pipeline, processorProvider);
+        }
     }
 
     @BeforeEach
     void setUp() {
         processors = List.of(processor);
+
+        when(pluginMetrics.counter(any())).thenReturn(counter);
     }
 
     @Nested
@@ -118,6 +134,8 @@ public void testProcessAcknowledgementsSuccess() {
             PipelineRunnerImpl pipelineRunner = createObjectUnderTest();
             pipelineRunner.processAcknowledgements(inputEvents, outputRecords);
             verify(defaultEventHandle).release(true);
+
+            verifyNoInteractions(counter);
         }
 
         @Test
@@ -134,6 +152,8 @@ void testProcessAcknowledgementsReleasesMissingEvents() {
             pipelineRunner.processAcknowledgements(inputEvents, outputRecords);
             verify(defaultEventHandle).release(true);
             assertNotSame(event, differentEvent);
+
+            verifyNoInteractions(counter);
         }
 
         @Test
@@ -148,6 +168,8 @@ void testProcessAcknowledgementsDoesNotReleaseWhenEventsPresent() {
             PipelineRunnerImpl pipelineRunner = createObjectUnderTest();
             pipelineRunner.processAcknowledgements(inputEvents, outputRecords);
             verify(defaultEventHandle, never()).release(true);
+
+            verifyNoInteractions(counter);
         }
 
         @Test
@@ -156,18 +178,20 @@ void testProcessAcknowledgementsInvalidEventHandleIncrementsCounter() {
             Collection<Record<Event>> outputRecords = List.of(record);
             when(event.getEventHandle()).thenReturn(eventHandle);
 
-            try (MockedStatic<PluginMetrics> pluginMetricsStatic = mockStatic(PluginMetrics.class)) {
-                final PluginMetrics pluginMetrics = mock(PluginMetrics.class);
-                final Counter counter = mock(Counter.class);
-                when(pluginMetrics.counter(any())).thenReturn(counter);
-                pluginMetricsStatic.when(() -> PluginMetrics.fromNames("PipelineRunner", pipeline.getName()))
-                        .thenReturn(pluginMetrics);
+            assertDoesNotThrow(() -> createObjectUnderTest().processAcknowledgements(inputEvents, outputRecords));
+            verify(counter, atLeastOnce()).increment();
+        }
 
-                PipelineRunnerImpl pipelineRunner = createObjectUnderTest();
-                assertDoesNotThrow(() -> pipelineRunner.processAcknowledgements(inputEvents, outputRecords));
+        @Test
+        void testProcessAcknowledgementsInvalidEventHandleIncrementsCounter2() {
+            Record<Event> eventRecord = mock(Record.class);
+            when(eventRecord.getData()).thenReturn(event);
+            List<Event> inputEvents = List.of(event);
+            Collection<Record<Event>> outputRecords = List.of(record, eventRecord);
+            when(event.getEventHandle()).thenReturn(eventHandle);
 
-                verify(counter, atLeastOnce()).increment();
-            }
+            assertDoesNotThrow(() -> createObjectUnderTest().processAcknowledgements(inputEvents, outputRecords));
+            verifyNoInteractions(counter);
         }
     }
 

data-prepper-plugins/aggregate-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/aggregate/AggregateGroup.java

@@ -28,27 +28,26 @@ class AggregateGroup implements AggregateActionInput {
     private final Lock handleEventForGroupLock;
     private final Map<Object, Object> identificationKeys;
     private Function<Duration, Boolean> customShouldConclude;
-    private EventHandle eventHandle;
+    private EventHandle groupEventHandle;
 
     AggregateGroup(final Map<Object, Object> identificationKeys) {
         this.groupState = new DefaultGroupState();
         this.identificationKeys = identificationKeys;
         this.groupStart = Instant.now();
         this.concludeGroupLock = new ReentrantLock();
         this.handleEventForGroupLock = new ReentrantLock();
-        this.eventHandle = new AggregateEventHandle(Instant.now());
+        this.groupEventHandle = new AggregateEventHandle(Instant.now());
     }
 
     @Override
     public EventHandle getEventHandle() {
-        return eventHandle;
+        return groupEventHandle;
     }
 
-    public void attachToEventAcknowledgementSet(Event event) {
-        InternalEventHandle internalEventHandle;
-        EventHandle handle = event.getEventHandle();
-        internalEventHandle = (InternalEventHandle)(handle);
-        internalEventHandle.addEventHandle(eventHandle);
+    public void attachToEventAcknowledgementSet(final Event event) {
+        final EventHandle handle = event.getEventHandle();
+        final InternalEventHandle internalEventHandle = (InternalEventHandle) (handle);
+        internalEventHandle.addEventHandle(groupEventHandle);
     }
 
     public GroupState getGroupState() {
@@ -86,6 +85,6 @@ boolean shouldConcludeGroup(final Duration groupDuration) {
     void resetGroup() {
         groupStart = Instant.now();
         groupState.clear();
-        this.eventHandle = new AggregateEventHandle(groupStart);
+        this.groupEventHandle = new AggregateEventHandle(groupStart);
     }
 }