@Override
    public void initTransport(OptionHolder sinkOptionHolder, List<OptionHolder> destinationOptionHolders,
                              Map<String, String> deploymentProperties,
                              List<Map<String, String>> destinationDeploymentProperties,
                              Annotation sinkAnnotation, ConfigReader sinkConfigReader,
                              DistributionStrategy strategy, String type, SiddhiAppContext siddhiAppContext) {
        final String transportType = sinkOptionHolder.validateAndGetStaticValue(SiddhiConstants
                .ANNOTATION_ELEMENT_TYPE);
        Extension sinkExtension = DefinitionParserHelper.constructExtension
                (streamDefinition, SiddhiConstants.ANNOTATION_SINK, transportType, sinkAnnotation, SiddhiConstants
                        .NAMESPACE_SINK);

        Set<String> allDynamicOptionKeys = findAllDynamicOptions(destinationOptionHolders);
        destinationOptionHolders.forEach(optionHolder -> {
            optionHolder.merge(sinkOptionHolder);
            allDynamicOptionKeys.forEach(optionKey -> {
                String optionValue = optionHolder.getOrCreateOption(optionKey, null).getValue();
                if (optionValue == null || optionValue.isEmpty()) {
                    throw new SiddhiAppValidationException("Destination properties can only contain " +
                            "non-empty static values.");
                }

                Option sinkOption = sinkOptionHolder.getOrAddStaticOption(optionKey, optionValue);
                sinkOption.addVariableValue(optionValue);
                destinationCount++;
            });
        });
        this.sink = (Sink) SiddhiClassLoader.loadExtensionImplementation(
                sinkExtension, SinkExecutorExtensionHolder.getInstance(siddhiAppContext));
        this.sink.initOnlyTransport(streamDefinition, sinkOptionHolder, sinkConfigReader, type,
                new SingleClientConnectionCallback(destinationCount, strategy), destinationDeploymentProperties.get(0),
                siddhiAppContext);
        if (!this.sink.getServiceDeploymentInfoList().isEmpty()) {
            ((ServiceDeploymentInfo) this.sink.getServiceDeploymentInfoList().get(0)).
                    addDeploymentProperties(deploymentProperties);
        }
    }
This class implements a the distributed sink that could publish to multiple destination using a single
client/publisher. Following are some examples,
- In a case where there are multiple partitions in a single topic in Kafka, the same kafka client can be used
to send events to all the partitions within the topic.
- The same email client can send email to different addresses.
modules/siddhi-core/src/main/java/io/siddhi/core/util/transport/SingleClientDistributedSink.java