Add acknowledgement support to aggregate processor #5139

kkondaka · 2024-10-31T22:27:48Z

Description

Add acknowledgement support to aggregate processor
Unaggregated events are not passed to the next processor in chain. For such events, DataPrepper core (in ProcessWorker) would send acknowledgements. It is possible for some aggregate actions (like TailSampler) to "hold" un-aggregated events. holdsEvents() processor API indicates that and DataPrepper code wouldn't send acks for such events.

Issues Resolved

Resolves #[Issue number to be closed when this PR is merged]

Check List

[X ] New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
[X ] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

dlvenable · 2024-12-12T22:28:16Z

...sor/src/main/java/org/opensearch/dataprepper/plugins/processor/aggregate/AggregateGroup.java

@@ -62,6 +82,7 @@ boolean shouldConcludeGroup(final Duration groupDuration) {

    void resetGroup() {
        groupStart = Instant.now();
+        this.eventHandle = new AggregateEventHandle(groupStart);


Should this do something with the old eventHandle? Release it? Something else?

No. The previous event handle was sent with the concluded event and will be released after the event is sent to Sink. Every "aggregate duration" ONE aggregated event is created and one aggregate event handle is created. The aggregate event handle is attached to the aggregated event and the group will create a new handle for the new duration. If the event handle is not attached to any event, then it will be freed (by garbage collector) when a new event handle is created in resetGroup.

Thanks for the explanation. This is clear. To be sure, are there any paths that could lead to resetting a group where we do need to handle the aggregate event?

It doesn't look like it. It looks like resetGroup() is called only in closeGroup() which is called only when the group is concluded.

dlvenable · 2024-12-12T22:33:59Z

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


Update this to be a little clear on what it means to hold events.

dlvenable

This looks good overall. I'm good once you make a few small tweaks such as documentation improvements and add tests.

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

David has approved changes before the test cases are added

graytaylor0 · 2024-12-18T22:33:06Z

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

@@ -125,6 +125,7 @@ public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
            }
            final IdentificationKeysHasher.IdentificationKeysMap identificationKeysMap = identificationKeysHasher.createIdentificationKeysMapFromEvent(event);
            final AggregateGroup aggregateGroupForEvent = aggregateGroupManager.getAggregateGroup(identificationKeysMap);
+            aggregateGroupForEvent.attachToEventAcknowledgementSet(event);


So the solution is to create a new event handle to represent the aggregate group? Makes sense but what is happening exactly to the original event's event handle. Those don't have to be released?

They are released by ProcessWorker. In the following code

records = processor.execute(records); if (inputEvents != null) { processAcknowledgements(inputEvents, records); }

where every event that's not passed to the next processor is acknowledged

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dinujoh · 2024-12-19T17:26:30Z

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

@@ -125,6 +125,7 @@ public Collection<Record<Event>> doExecute(Collection<Record<Event>> records) {
            }
            final IdentificationKeysHasher.IdentificationKeysMap identificationKeysMap = identificationKeysHasher.createIdentificationKeysMapFromEvent(event);
            final AggregateGroup aggregateGroupForEvent = aggregateGroupManager.getAggregateGroup(identificationKeysMap);


Consider automating the attachment of an event to the AggregateGroup during object creation so that it's encapsulated within the AggregateGroupManager

public AggregateGroup getAggregateGroup(Map<Object, Object> identificationKeysMap, Event event) { AggregateGroup aggregateGroup = getOrCreateAggregateGroup(identificationKeysMap); ... aggregateGroup.attachToEventAcknowledgementSet(event); return aggregateGroup; }

dinujoh · 2024-12-19T17:29:11Z

...g/opensearch/dataprepper/plugins/processor/aggregate/actions/RateLimiterAggregateAction.java

+    public AggregateActionOutput concludeGroup(final AggregateActionInput aggregateActionInput) {
+        if (aggregateActionInput != null) {
+            EventHandle eventHandle = aggregateActionInput.getEventHandle();
+            if (eventHandle != null) {
+                eventHandle.release(true);
+            }
+        }
+        return new AggregateActionOutput(Collections.emptyList());
+    }


Can this be default implementation in the interface ? I see the same logic for multiple implementation.

That's a good suggestion.

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dinujoh · 2024-12-19T23:34:02Z

.../main/java/org/opensearch/dataprepper/plugins/processor/aggregate/AggregateGroupManager.java

@@ -26,6 +27,12 @@ AggregateGroup getAggregateGroup(final IdentificationKeysHasher.IdentificationKe
        return allGroups.computeIfAbsent(identificationKeysMap, (hash) -> new AggregateGroup(identificationKeysMap.getKeyMap()));
    }

+    AggregateGroup getAggregateGroupForEvent(final IdentificationKeysHasher.IdentificationKeysMap identificationKeysMap, final Event event) {
+        AggregateGroup aggregateGroup = getAggregateGroup(identificationKeysMap);
+        aggregateGroup.attachToEventAcknowledgementSet(event);


we are repeating this in both AggregateGroupManager and AggregateProcessor line 98

Forgot to remove this.

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

* Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments and added tests Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Fixed checkstyle errors Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Fixed test errors by adding await Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Addressed review comments Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Removed unnecessary API Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> * Fixed checkstyle error Signed-off-by: Krishna Kondaka <krishkdk@amazon.com> --------- Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dlvenable previously requested changes Dec 12, 2024

View reviewed changes

dlvenable marked this pull request as ready for review December 13, 2024 19:21

dlvenable requested review from sb2k16, chenqi0805, engechas, graytaylor0, dinujoh, KarstenSchnitter and oeyh as code owners December 13, 2024 19:21

dlvenable reviewed Dec 13, 2024

View reviewed changes

Addressed review comments

3f2d59f

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

kkondaka force-pushed the aggr-acks branch from 189cdf9 to 3f2d59f Compare December 13, 2024 20:52

kkondaka added 2 commits December 18, 2024 07:32

Addressed review comments and added tests

2ac8584

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

Fixed checkstyle errors

98d0644

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

graytaylor0 previously approved these changes Dec 18, 2024

View reviewed changes

Fixed test errors by adding await

7cb44d2

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

kkondaka dismissed graytaylor0’s stale review via 7cb44d2 December 18, 2024 23:10

graytaylor0 previously approved these changes Dec 19, 2024

View reviewed changes

dinujoh reviewed Dec 19, 2024

View reviewed changes

Addressed review comments

2af6722

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

kkondaka dismissed graytaylor0’s stale review via 2af6722 December 19, 2024 19:49

graytaylor0 previously approved these changes Dec 19, 2024

View reviewed changes

dinujoh reviewed Dec 19, 2024

View reviewed changes

Removed unnecessary API

47ada91

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

kkondaka dismissed graytaylor0’s stale review via 47ada91 December 19, 2024 23:51

Fixed checkstyle error

d78769d

Signed-off-by: Krishna Kondaka <krishkdk@amazon.com>

dinujoh approved these changes Dec 20, 2024

View reviewed changes

kkondaka merged commit 708843c into opensearch-project:main Dec 20, 2024
47 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add acknowledgement support to aggregate processor #5139

Add acknowledgement support to aggregate processor #5139

kkondaka commented Oct 31, 2024 •

edited

Loading

dlvenable Dec 12, 2024

kkondaka Dec 12, 2024 •

edited

Loading

dlvenable Dec 13, 2024

kkondaka Dec 13, 2024

dlvenable Dec 12, 2024

dlvenable left a comment

graytaylor0 Dec 18, 2024

kkondaka Dec 18, 2024

dinujoh Dec 19, 2024

dinujoh Dec 19, 2024 •

edited

Loading

kkondaka Dec 19, 2024

dinujoh Dec 19, 2024

kkondaka Dec 19, 2024

Add acknowledgement support to aggregate processor #5139

Add acknowledgement support to aggregate processor #5139

Conversation

kkondaka commented Oct 31, 2024 • edited Loading

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

kkondaka Dec 12, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dinujoh Dec 19, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kkondaka commented Oct 31, 2024 •

edited

Loading

kkondaka Dec 12, 2024 •

edited

Loading

dinujoh Dec 19, 2024 •

edited

Loading