Adds additional thread synchronization in the AggregateProcessor to prevent duplicate or orphaned aggregate groups. (#6439)

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

Author: dlvenable

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

@@ -105,21 +105,23 @@ private AggregateGroup getAggregateGroupForEvent(final IdentificationKeysHasher.
     public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
         final List<Record<Event>> recordsOut = new LinkedList<>();
 
-        final List<Map.Entry<IdentificationKeysHasher.IdentificationKeysMap, AggregateGroup>> groupsToConclude = aggregateGroupManager.getGroupsToConclude(forceConclude);
-        for (final Map.Entry<IdentificationKeysHasher.IdentificationKeysMap, AggregateGroup> groupEntry : groupsToConclude) {
-            final AggregateActionOutput actionOutput = aggregateActionSynchronizer.concludeGroup(groupEntry.getKey(), groupEntry.getValue(), forceConclude);
-
-            final List<Event> concludeGroupEvents = actionOutput != null ? actionOutput.getEvents() : null;
-            if (concludeGroupEvents != null && !concludeGroupEvents.isEmpty()) {
-                concludeGroupEvents.stream().forEach((event) -> {
-                    if (aggregatedEventsTag != null) {
-                        event.getMetadata().addTags(List.of(aggregatedEventsTag));
-                    }
-                    recordsOut.add(new Record(event));
-                    actionConcludeGroupEventsOutCounter.increment();
-                });
-            } else {
-                actionConcludeGroupEventsDroppedCounter.increment();
+        synchronized (this) {
+            final List<Map.Entry<IdentificationKeysHasher.IdentificationKeysMap, AggregateGroup>> groupsToConclude = aggregateGroupManager.getGroupsToConclude(forceConclude);
+            for (final Map.Entry<IdentificationKeysHasher.IdentificationKeysMap, AggregateGroup> groupEntry : groupsToConclude) {
+                final AggregateActionOutput actionOutput = aggregateActionSynchronizer.concludeGroup(groupEntry.getKey(), groupEntry.getValue(), forceConclude);
+
+                final List<Event> concludeGroupEvents = actionOutput != null ? actionOutput.getEvents() : null;
+                if (concludeGroupEvents != null && !concludeGroupEvents.isEmpty()) {
+                    concludeGroupEvents.stream().forEach((event) -> {
+                        if (aggregatedEventsTag != null) {
+                            event.getMetadata().addTags(List.of(aggregatedEventsTag));
+                        }
+                        recordsOut.add(new Record(event));
+                        actionConcludeGroupEventsOutCounter.increment();
+                    });
+                } else {
+                    actionConcludeGroupEventsDroppedCounter.increment();
+                }
             }
         }
 
@@ -132,9 +134,12 @@ public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
                 continue;
             }
             final IdentificationKeysHasher.IdentificationKeysMap identificationKeysMap = identificationKeysHasher.createIdentificationKeysMapFromEvent(event);
-            final AggregateGroup aggregateGroupForEvent = getAggregateGroupForEvent(identificationKeysMap);
+            final AggregateActionResponse handleEventResponse;
+            synchronized (this) {
+                final AggregateGroup aggregateGroupForEvent = getAggregateGroupForEvent(identificationKeysMap);
 
-            final AggregateActionResponse handleEventResponse = aggregateActionSynchronizer.handleEventForGroup(event, identificationKeysMap, aggregateGroupForEvent);
+                handleEventResponse = aggregateActionSynchronizer.handleEventForGroup(event, identificationKeysMap, aggregateGroupForEvent);
+            }
 
             final Event aggregateActionResponseEvent = handleEventResponse.getEvent();