Load AggregateAction, implement AggregateAction.handleEvent in AggregateProcessor, add GroupStateManager and AggregateIdentificationKeysHasher classes for use in AggregateProcessor

data-prepper-plugins/aggregate-processor/build.gradle

@@ -7,12 +7,21 @@ plugins {
     id 'java'
 }
 
-repositories {
-    mavenCentral()
-}
-
 dependencies {
     implementation project(':data-prepper-api')
     implementation project(':data-prepper-plugins:common')
     implementation 'com.fasterxml.jackson.core:jackson-databind'
+    testImplementation 'org.hamcrest:hamcrest:2.2'
+    testImplementation "org.mockito:mockito-inline:${versionMap.mockito}"
+}
+
+jacocoTestCoverageVerification {
+    dependsOn jacocoTestReport
+    violationRules {
+        rule {
+            limit {
+                minimum = 1.0
+            }
+        }
+    }
 }

data-prepper-plugins/aggregate-processor/src/main/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateActionResponse.java

@@ -8,12 +8,11 @@
 import com.amazon.dataprepper.model.event.Event;
 
 /**
- * Model class to be returned in {@link com.amazon.dataprepper.plugins.processor.aggregate.AggregateAction}. Contains both the Event to be processed and an option
- * to close the current window for {@link com.amazon.dataprepper.plugins.processor.aggregate.AggregateProcessor} immediately after an event is handled.
+ * Model class to be returned in {@link com.amazon.dataprepper.plugins.processor.aggregate.AggregateAction}. Contains the Event to be processed, which is null if no Event should be processed.
  * @since 1.3
  */
 public class AggregateActionResponse {
-    private Event event;
+    private final Event event;
 
     public AggregateActionResponse(final Event event) {
         this.event = event;

data-prepper-plugins/aggregate-processor/src/main/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateGroup.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugins.processor.aggregate;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class AggregateGroup {
+    private final Map<Object, Object> groupState = new HashMap<>();
+
+    AggregateGroup() {
+
+    }
+
+    public Map<Object, Object> getGroupState() {
+        return groupState;
+    }
+}

data-prepper-plugins/aggregate-processor/src/main/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateGroupManager.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugins.processor.aggregate;
+
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+
+class AggregateGroupManager {
+
+    private final Map<AggregateIdentificationKeysHasher.IdentificationHash, AggregateGroup> allGroupStates = Maps.newConcurrentMap();
+
+    AggregateGroupManager() {
+
+    }
+
+    AggregateGroup getAggregateGroup(final AggregateIdentificationKeysHasher.IdentificationHash identificationHash) {
+        final AggregateGroup aggregateGroup = allGroupStates.get(identificationHash);
+
+        if (aggregateGroup != null) {
+            return aggregateGroup;
+        }
+
+        final AggregateGroup newAggregateGroup = new AggregateGroup();
+        allGroupStates.put(identificationHash, newAggregateGroup);
+        return newAggregateGroup;
+    }
+}

data-prepper-plugins/aggregate-processor/src/main/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateIdentificationKeysHasher.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugins.processor.aggregate;
+
+import com.amazon.dataprepper.model.event.Event;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+class AggregateIdentificationKeysHasher {
+    private final List<String> identificationKeys;
+    AggregateIdentificationKeysHasher(final List<String> identificationKeys) {
+        this.identificationKeys = identificationKeys;
+    }
+
+    IdentificationHash createIdentificationKeyHashFromEvent(final Event event) {
+        final Map<Object, Object> identificationKeysHash = new HashMap<>();
+        for (final String identificationKey : identificationKeys) {
+            identificationKeysHash.put(identificationKey, event.get(identificationKey, Object.class));
+        }
+        return new IdentificationHash(identificationKeysHash);
+    }
+
+    public static class IdentificationHash {
+        private final Map<Object, Object> hash;
+
+        IdentificationHash(final Map<Object, Object> hash) {
+            this.hash = hash;
+        }
+
+        @Override
+        public boolean equals(Object other) {
+            return this.hash.equals(other);
+        }
+
+        @Override
+        public int hashCode() {
+            return hash.hashCode();
+        }
+    }
+}

data-prepper-plugins/aggregate-processor/src/main/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateProcessor.java

@@ -8,8 +8,8 @@
 import com.amazon.dataprepper.metrics.PluginMetrics;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.annotations.DataPrepperPluginConstructor;
-import com.amazon.dataprepper.model.annotations.SingleThread;
-import com.amazon.dataprepper.model.configuration.PipelineDescription;
+import com.amazon.dataprepper.model.configuration.PluginModel;
+import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.event.Event;
 import com.amazon.dataprepper.model.plugin.PluginFactory;
 import com.amazon.dataprepper.model.processor.AbstractProcessor;
@@ -19,23 +19,54 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
+import java.util.LinkedList;
+import java.util.List;
 
-@SingleThread
 @DataPrepperPlugin(name = "aggregate", pluginType = Processor.class, pluginConfigurationType = AggregateProcessorConfig.class)
 public class AggregateProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
     private static final Logger LOG = LoggerFactory.getLogger(AggregateProcessor.class);
 
     private final AggregateProcessorConfig aggregateProcessorConfig;
+    private final AggregateGroupManager aggregateGroupManager;
+    private final AggregateAction aggregateAction;
+    private final AggregateIdentificationKeysHasher aggregateIdentificationKeysHasher;
 
     @DataPrepperPluginConstructor
-    public AggregateProcessor(final AggregateProcessorConfig aggregateProcessorConfig, final PluginMetrics pluginMetrics, final PluginFactory pluginFactory, final PipelineDescription pipelineDescription) {
+    public AggregateProcessor(final AggregateProcessorConfig aggregateProcessorConfig, final PluginMetrics pluginMetrics, final PluginFactory pluginFactory) {
+        this(aggregateProcessorConfig, pluginMetrics, pluginFactory, new AggregateGroupManager(), new AggregateIdentificationKeysHasher(aggregateProcessorConfig.getIdentificationKeys()));
+    }
+    public AggregateProcessor(final AggregateProcessorConfig aggregateProcessorConfig, final PluginMetrics pluginMetrics, final PluginFactory pluginFactory, final AggregateGroupManager aggregateGroupManager, final AggregateIdentificationKeysHasher aggregateIdentificationKeysHasher) {
         super(pluginMetrics);
         this.aggregateProcessorConfig = aggregateProcessorConfig;
+        this.aggregateGroupManager = aggregateGroupManager;
+        this.aggregateIdentificationKeysHasher = aggregateIdentificationKeysHasher;
+        this.aggregateAction = loadAggregateAction(pluginFactory);
+    }
+
+    private AggregateAction loadAggregateAction(final PluginFactory pluginFactory) {
+        final PluginModel actionConfiguration = aggregateProcessorConfig.getAggregateAction();
+        final PluginSetting actionPluginSetting = new PluginSetting(actionConfiguration.getPluginName(), actionConfiguration.getPluginSettings());
+        return pluginFactory.loadPlugin(AggregateAction.class, actionPluginSetting);
     }
 
     @Override
     public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
-        return records;
+        final List<Record<Event>> recordsOut = new LinkedList<>();
+
+        for (final Record<Event> record : records) {
+            final Event event = record.getData();
+            final AggregateIdentificationKeysHasher.IdentificationHash identificationKeysHash = aggregateIdentificationKeysHasher.createIdentificationKeyHashFromEvent(event);
+            final AggregateGroup aggregateGroupForEvent = aggregateGroupManager.getAggregateGroup(identificationKeysHash);
+
+            final AggregateActionResponse handleEventResponse = aggregateAction.handleEvent(event, aggregateGroupForEvent.getGroupState());
+
+            final Event aggregateActionResponseEvent = handleEventResponse.getEvent();
+
+            if (aggregateActionResponseEvent != null) {
+                recordsOut.add(new Record<>(aggregateActionResponseEvent, record.getMetadata()));
+            }
+        }
+        return recordsOut;
     }
 
     @Override

data-prepper-plugins/aggregate-processor/src/main/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateProcessorConfig.java

@@ -5,11 +5,13 @@
 
 package com.amazon.dataprepper.plugins.processor.aggregate;
 
+import com.amazon.dataprepper.model.configuration.PluginModel;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import jakarta.validation.constraints.AssertTrue;
 import jakarta.validation.constraints.Min;
 import jakarta.validation.constraints.NotEmpty;
+import jakarta.validation.constraints.NotNull;
 
 import java.io.File;
 import java.util.List;
@@ -31,6 +33,10 @@ public class AggregateProcessorConfig {
     @NotEmpty
     private String dbPath = DEFAULT_DB_PATH;
 
+    @JsonProperty("action")
+    @NotNull
+    private PluginModel aggregateAction;
+
     @JsonIgnore
     private File dbFile;
 
@@ -46,6 +52,8 @@ public String getDbPath() {
         return dbPath;
     }
 
+    public PluginModel getAggregateAction() { return aggregateAction; }
+
     @AssertTrue(message = "db_path is not a valid file path")
     boolean isDbPathValid() {
         dbFile = new File(dbPath);

data-prepper-plugins/aggregate-processor/src/test/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateGroupManagerTest.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugins.processor.aggregate;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.CoreMatchers.sameInstance;
+
+public class AggregateGroupManagerTest {
+
+    private AggregateGroupManager aggregateGroupManager;
+
+    private AggregateIdentificationKeysHasher.IdentificationHash identificationHash;
+
+    @BeforeEach
+    void setup() {
+        final Map<Object, Object> identificationKeysHash = new HashMap<>();
+        identificationKeysHash.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+
+        identificationHash = new AggregateIdentificationKeysHasher.IdentificationHash(identificationKeysHash);
+    }
+
+    private AggregateGroupManager createObjectUnderTest() {
+        return new AggregateGroupManager();
+    }
+
+    @Test
+    void getGroupState_with_non_existing_group_state_creates_and_returns_new_group_state_and_adds_to_allGroupStates() {
+        aggregateGroupManager = createObjectUnderTest();
+
+        final AggregateGroup emptyAggregateGroup = aggregateGroupManager.getAggregateGroup(identificationHash);
+        assertThat(emptyAggregateGroup, notNullValue());
+        assertThat(emptyAggregateGroup.getGroupState(), equalTo(Collections.emptyMap()));
+
+        final AggregateGroup secondAggregateGroup = aggregateGroupManager.getAggregateGroup(identificationHash);
+        assertThat(secondAggregateGroup, notNullValue());
+        assertThat(secondAggregateGroup, is(sameInstance(emptyAggregateGroup)));
+    }
+
+    @Test
+    void getGroupState_returns_a_mutable_GroupState_Map() {
+        aggregateGroupManager = createObjectUnderTest();
+
+        final AggregateGroup firstAggregateGroup = aggregateGroupManager.getAggregateGroup(identificationHash);
+        firstAggregateGroup.getGroupState().put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+
+        final AggregateGroup secondAggregateGroup = aggregateGroupManager.getAggregateGroup(identificationHash);
+        assertThat(secondAggregateGroup, equalTo(firstAggregateGroup));
+
+    }
+}

data-prepper-plugins/aggregate-processor/src/test/java/com/amazon/dataprepper/plugins/processor/aggregate/AggregateIdentificationKeysHasherTest.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugins.processor.aggregate;
+
+import com.amazon.dataprepper.model.event.Event;
+import com.amazon.dataprepper.model.event.JacksonEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class AggregateIdentificationKeysHasherTest {
+    private Event event;
+    private List<String> identificationKeys;
+    private AggregateIdentificationKeysHasher aggregateIdentificationKeysHasher;
+
+    @BeforeEach
+    void setup() {
+        identificationKeys = new ArrayList<>();
+        identificationKeys.add("firstIdentificationKey");
+        identificationKeys.add("secondIdentificationKey");
+    }
+
+    private AggregateIdentificationKeysHasher createObjectUnderTest() {
+        return new AggregateIdentificationKeysHasher(identificationKeys);
+    }
+
+    @Test
+    void createIdentificationKeyHashFromEvent_returns_expected_Map() {
+        aggregateIdentificationKeysHasher = createObjectUnderTest();
+        final Map<Object, Object> eventMap = new HashMap<>();
+        eventMap.put("firstIdentificationKey", UUID.randomUUID().toString());
+        eventMap.put("secondIdentificationKey", UUID.randomUUID().toString());
+
+        final Map<Object, Object> expectedResult = new HashMap<>(eventMap);
+
+        eventMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+
+        event = JacksonEvent.builder()
+                .withEventType("event")
+                .withData(eventMap)
+                .build();
+
+        final AggregateIdentificationKeysHasher.IdentificationHash result = aggregateIdentificationKeysHasher.createIdentificationKeyHashFromEvent(event);
+        assertThat(result, equalTo(expectedResult));
+    }
+
+    @Test
+    void createIdentificationKeysHashFromEvent_where_Event_does_not_contain_one_of_the_identification_keys_returns_expected_Map() {
+        aggregateIdentificationKeysHasher = createObjectUnderTest();
+        final Map<Object, Object> eventMap = new HashMap<>();
+        eventMap.put("firstIdentificationKey", UUID.randomUUID().toString());
+
+        final Map<Object, Object> expectedResult = new HashMap<>(eventMap);
+        expectedResult.put("secondIdentificationKey", null);
+
+        eventMap.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
+
+        event = JacksonEvent.builder()
+                .withEventType("event")
+                .withData(eventMap)
+                .build();
+
+        final AggregateIdentificationKeysHasher.IdentificationHash result = aggregateIdentificationKeysHasher.createIdentificationKeyHashFromEvent(event);
+        assertThat(result, equalTo(expectedResult));
+    }
+}