opensearch-project / opensearch-project/data-prepper
Add acknowledgements support to aggregate processor
@kkondaka is already working on this.
Since Jan 23, 2024.
- Dominant language
- Java
- Stars
- 374
- Forks
- 354
- Avg merge
- 3d 18h
- Merged PRs (30d)
- 8
Description
Is your feature request related to a problem? Please describe.
End to end acknowledgements in DataPrepper do not support aggregate processor because the events may be forwarded to a remote peer. But if the local aggregation is used (discovery_mode: local_node in peer forwarder configuration), the events are not forwarded to remote peer and end-to-end acknowledgements may be supported in that case.
For example, if a aggregate_processor is configured with an aggregation action, and N number of events are sent to the processor, the individual events are not sent to the sink and acknowledgements for each of the individual events is sent back to the source. And source considers the data is durable even though the aggregated event itself is not sent to the sink (yet). And it is possible that due to some unexpected event after conclusion of the aggregation period, the DataPrepper fails to send aggregated event to the sink.
Describe the solution you'd like
Solution is to create an aggregate event handle and register it with all AcknowledgementSets that contribute to aggregate event (before it is concluded). Upon the conclusion of the aggregation, if there are no aggregated events generated, release the aggregated event handle. If any events are generated, associate the event handle with all events generated.
Describe alternatives you've considered (Optional)
A clear and concise description of any alternative solutions or features you've considered.
Additional context
Add any other context or screenshots about the feature request here.
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Assessment
This issue has not been assessed yet.