Allow peer forwarder to skip sending events to remote peer

[kkondaka]

Is your feature request related to a problem? Please describe.
Peer Forwarder sends some events to remote peer based on the hash function. If the inner processor of Peer forwarder has "when" condition, it is possible an event is forwarded to remote peer first and then get dropped because when condition evaluates to false. This is very sub-optimal. Also in some cases an option to force local aggregation may be needed.

Describe the solution you'd like
Add a new API to RequiresPeerForwarding.java some thing like

Boolean shouldForwardToRemotePeer(final Event event);   

This will allow innerProcessor to evaluate when condition and also check if localOnly option is configured.

For example, aggregate processor could implement the new API as follows

@Override                                                                                                                                           
    public Boolean shouldForwardToRemotePeer(Event event) {                                                                                             
        if (localOnly || (whenCondition != null && !expressionEvaluator.evaluateConditional(whenCondition, event))) {                                   
            return false;                                                                                                                               
        }                                                                                                                                               
        return true;                                                                                                                                    
    }   

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.

[dlvenable]

@kkondaka ,

I like the overall proposal. A few things I'd suggest changing:

1. Make this operate on a collection, similar to how the process method takes a collection of input events and provides output events.
2. The current name indicates that it should forward. But, the decision is whether or not peer forwarding is needed at all. An event may require peer forwarding, but be on the current node already. Maybe rename to applicableEventsForPeerForwarding.
3. We can make this have a default implementation of returning all events.

default Collection<Record<Event>> applicableEventsForPeerForwarding(Collection<Record<Event>> records) {
  return records;
}

What is the localOnly option? I believe if you do not define the peer_forwarder in Data Prepper configuration, this code is already handled by the core framework.

[dlvenable]

I took another look. To disable peer-forwarding, you set the following in the data-prepper-config.yaml.

peer_forwarder:
  discovery_mode: local_node

[dlvenable]

Completed by #4004