Added holdEvents API to processor

Signed-off-by: Kondaka <[email protected]>

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/processor/Processor.java

@@ -26,6 +26,14 @@ public interface Processor<InputRecord extends Record<?>, OutputRecord extends R
      */
     Collection<OutputRecord> execute(Collection<InputRecord> records);
 
+    /**
+     * indicates if the processor holds the events or not
+     */
+    default boolean holdsEvents() {
+        return false;
+    }
+
+
     /**
      * @since 1.2
      * Indicates to the processor that shutdown is imminent and any data currently held by the Processor

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

@@ -137,7 +137,8 @@ private void doRun() {
 
             try {
                 records = processor.execute(records);
-                if (inputEvents != null) {
+                // acknowledge missing events only if the processor is not holding events
+                if (!processor.holdsEvents() && inputEvents != null) {
                     processAcknowledgements(inputEvents, records);
                 }
             } catch (final Exception e) {

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

@@ -41,4 +41,12 @@ default AggregateActionOutput concludeGroup(final AggregateActionInput aggregate
         return new AggregateActionOutput(Collections.emptyList());
     }
 
+    /**
+     * indicates if the action holds the events or not
+     *
+     */
+    default boolean holdsEvents() {
+        return false;
+    }
+
 }

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

@@ -189,6 +189,11 @@ public void shutdown() {
 
     }
 
+    @Override
+    public boolean holdsEvents() {
+        return aggregateAction.holdsEvents();
+    }
+
     @Override
     public boolean isForLocalProcessingOnly(Event event) {
         // no need to check for when condition here because it is

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

@@ -35,6 +35,7 @@ public RateLimiterAggregateAction(final RateLimiterAggregateActionConfig ratelim
     public AggregateActionResponse handleEvent(final Event event, final AggregateActionInput aggregateActionInput) {
         if (rateLimiterMode == RateLimiterMode.DROP) {
             if (!rateLimiter.tryAcquire()) {
+                event.getEventHandle().release(true);
                 return AggregateActionResponse.nullEventResponse();
             }
         } else {

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

@@ -62,7 +62,6 @@ public AggregateActionResponse handleEvent(final Event event, final AggregateAct
         }
         List<Event> events = (List)groupState.getOrDefault(EVENTS_KEY, new ArrayList<>());
         events.add(event);
-	// TODO: This event shouldn't be acknowledged by data prepper core
         groupState.put(EVENTS_KEY, events);
         if (condition != null && !condition.isEmpty() && expressionEvaluator.evaluateConditional(condition, event)) {
             groupState.put(ERROR_STATUS_KEY, true);
@@ -78,6 +77,10 @@ public AggregateActionOutput concludeGroup(final AggregateActionInput aggregateA
         if (((groupState.containsKey(ERROR_STATUS_KEY) && (Boolean)groupState.get(ERROR_STATUS_KEY) == true)) || (randomInt < percent)) {
             return new AggregateActionOutput((List)groupState.getOrDefault(EVENTS_KEY, List.of()));
         }
+        List<Event> events = (List)groupState.getOrDefault(EVENTS_KEY, List.of());
+        for (final Event event : events) {
+            event.getEventHandle().release(true);
+        }
         return new AggregateActionOutput(List.of());
     }