[BUG] Unable to create stateful processors with multiple workers.

[dlvenable]

Describe the bug

Creating a stateful processor that requires peer forwarding does not work with multiple workers.

2024-06-25T00:02:23,474 [main] ERROR org.opensearch.dataprepper.parser.PipelineTransformer - Construction of pipeline components failed, skipping building of pipeline [service-map-pipeline] and its connected pipelines
java.lang.RuntimeException: Data Prepper 2.0 will only support a single peer-forwarder per pipeline/plugin type
	at org.opensearch.dataprepper.peerforwarder.DefaultPeerForwarderProvider.register(DefaultPeerForwarderProvider.java:42) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.peerforwarder.LocalModePeerForwarderProvider.register(LocalModePeerForwarderProvider.java:32) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.peerforwarder.PeerForwardingProcessorDecorator.lambda$decorateProcessors$1(PeerForwardingProcessorDecorator.java:73) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:?]
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
	at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
	at org.opensearch.dataprepper.peerforwarder.PeerForwardingProcessorDecorator.decorateProcessors(PeerForwardingProcessorDecorator.java:76) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.parser.PipelineTransformer.lambda$buildPipelineFromConfiguration$3(PipelineTransformer.java:132) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:?]
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197) ~[?:?]
	at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1625) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499) ~[?:?]
	at java.base/java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:921) ~[?:?]
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[?:?]
	at java.base/java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:682) ~[?:?]
	at org.opensearch.dataprepper.parser.PipelineTransformer.buildPipelineFromConfiguration(PipelineTransformer.java:137) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.parser.PipelineTransformer.transformConfiguration(PipelineTransformer.java:99) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:?]
	at org.opensearch.dataprepper.DataPrepper.<init>(DataPrepper.java:69) ~[data-prepper-core-2.9.0-SNAPSHOT.jar:2.9.0-SNAPSHOT]

To Reproduce

1. Edit trace_analytics_no_ssl_2x.yml by adding workers: 2 to the service-map-pipeline pipeline.
2. Go to examples/jaeger-hotrod
3. docker compose up -d
4. `docker logs -f data-prepper

Expected behavior

This should work just like it used to.

Environment (please complete the following information):

Data Prepper running from main.

[dlvenable]

I believe this was introduced by #4574.

https://github.com/opensearch-project/data-prepper/pull/4574/files#diff-82ab126c928e8f6c33e33fcdca6015c436d7df3870b9b420880f669cce2c8681R71-R76

I think the solution will be to re-use the existing peer-forwarder when the Processor instance is the same.