Support multiple peer-forwarding components in the same pipeline

[dlvenable]

Is your feature request related to a problem? Please describe.

Currently Data Prepper does not support multiple instances of processors that require peer-forwarding in a single pipeline.

processor:
  - aggregate:
  - aggregate:

This is primarily because Data Prepper does not currently distinguish between different plugins of the same type. So when forwarding events, the event would go back to the first aggregate processor.

data-prepper/data-prepper-core/src/main/java/org/opensearch/dataprepper/peerforwarder/PeerForwarderProvider.java

Lines 39 to 42 in db33254

	if (pipelinePeerForwarderReceiveBufferMap.containsKey(pipelineName) &&
	pipelinePeerForwarderReceiveBufferMap.get(pipelineName).containsKey(pluginId)) {
	throw new RuntimeException("Data Prepper 2.0 will only support a single peer-forwarder per pipeline/plugin type");
	}

There is also currently a restriction on sharing the same identification keys.

data-prepper/data-prepper-core/src/main/java/org/opensearch/dataprepper/peerforwarder/PeerForwardingProcessorDecorator.java

Lines 54 to 62 in 6d74ec4

	processors.forEach(processor -> {
	if (processor instanceof RequiresPeerForwarding) {
	final Set<String> processorIdentificationKeys = new HashSet<>(((RequiresPeerForwarding) processor).getIdentificationKeys());
	if (!identificationKeys.equals(processorIdentificationKeys)) {
	throw new RuntimeException(
	"All the processors of same type within a single pipeline should have same identification keys.");
	}
	}
	});

Describe the solution you'd like

Open up this functionality.

When routing events between peers, Data Prepper includes a destinationPluginId property.

data-prepper/data-prepper-core/src/main/java/org/opensearch/dataprepper/peerforwarder/model/PeerForwardingEvents.java

Line 21 in 5bc6a2c

private String destinationPluginId;

Currently, pluginIds are not unique. Thus, if we forwarded events Data Prepper would not know which aggregate processor to use - the first or second. We should update Data Prepper by:

1. Supporting dynamic pluginIds as noted in Pipeline Component Ids #1025 and Auto-generate unique component Ids #3995. We do not need configurable plugin Ids for this.
2. When forwarding events, use the pluginId instead of the plugin type.

By generating component Ids such as aggregate and aggregate2, Data Prepper can forward events to peers and target the correct processor in the processor chain.

Tasks

• Auto-generate unique component Ids #3995
• Investigate the identification keys restriction and solution
• Disable the restriction on the number of processors of the same type