Provide a way to terminate an aggregation group early in the aggregation processor

Is your feature request related to a problem? Please describe.
I have a pipeline to ingest logs in opensearch, and I use the aggregation processor with the put_map action.
At the moment, the only way a group can close with this action is to wait for the group_duration to expire.

That means that all records that have been merged but whose group is not yet closed still lives in memory in the data-prepper nodes.
For high throughput or high latency pipeline where you have to specify a large group_duration, or both, that means a lot of memory will be wasted on already merged records that are just waiting for the expiration of the group.

There should be a way to terminate a group and flush the result to the next processor or sink if you know you do not need to wait.

Describe the solution you'd like
The solution could work in two steps:

1. Configure an expression to evaluate when the aggregation action is executed. This expression would evaluate to a tag added to the metadata of the aggregation group for instance.
2. Attach an expression to evaluate or a list of tags that must be present on the aggregation group when it is mutated, that if true marks the group for finalization and flushes it immediately.

The pipeline configuration could look like:

my_pipeline:
  source:
    file:
      path: somefile.log
  processors:
    - aggregate:
         action:
           put_all: {}
         identification_keys:
           - common_key
         tag_on_aggregate: /log_type
         terminate_when: hasTags("type_1", "type_2")

Describe alternatives you've considered (Optional)

Other option: add a close_when expression common to all AggregateAction that provides the custom expression that guards the closure of the group.
This expression can be evaluated when AggregateGroupManager.getGroupsToConclude() is called, so the changes in AggregateProcessor are minimal.

Additional context

The aggregate processor first checks for groups to conclude and then processes the current batch. This logic should be reversed so the events are flushed immediately after the aggregation.

Start by reading the aggregation processor flow, especially AggregateGroupManager.getGroupsToConclude() and the ordering in AggregateProcessor. Compare the proposed terminate_when and close_when approaches, then define how aggregation actions mark groups and when those groups flush; done means an aggregation group can be finalized before group_duration expires without retaining already-merged records.