public static void addEventSink(StreamDefinition streamDefinition,
                                    StreamJunction streamJunction,
                                    ConcurrentMap<String, List<Sink>> eventSinkMap,
                                    SiddhiAppContext siddhiAppContext) {
        for (Annotation sinkAnnotation : streamDefinition.getAnnotations()) {
            if (SiddhiConstants.ANNOTATION_SINK.equalsIgnoreCase(sinkAnnotation.getName())) {
                try {
                    sinkAnnotation = updateAnnotationRef(sinkAnnotation, SiddhiConstants.NAMESPACE_SINK,
                            siddhiAppContext);
                    Annotation mapAnnotation = AnnotationHelper.getAnnotation(SiddhiConstants.ANNOTATION_MAP,
                            sinkAnnotation.getAnnotations());
                    String sinkType = sinkAnnotation.getElement(SiddhiConstants.ANNOTATION_ELEMENT_TYPE);
                    if (sinkType == null) {
                        throw new SiddhiAppCreationException(
                                "Attribute 'type' does not exist for annotation '" + sinkAnnotation + "'",
                                sinkAnnotation, siddhiAppContext);
                    }
                    if (mapAnnotation == null) {
                        mapAnnotation = Annotation.annotation(SiddhiConstants.ANNOTATION_MAP).element(SiddhiConstants
                                .ANNOTATION_ELEMENT_TYPE, "passThrough");
                    }
                    Annotation distributionAnnotation =
                            AnnotationHelper.getAnnotation(SiddhiConstants.ANNOTATION_DISTRIBUTION,
                                    sinkAnnotation.getAnnotations());

                    if (mapAnnotation != null) {
                        String[] supportedDynamicOptions = null;
                        List<OptionHolder> destinationOptHolders = new ArrayList<>();
                        List<Map<String, String>> destinationDeploymentProperties = new ArrayList<>();
                        Extension sinkExtension = constructExtension(streamDefinition, SiddhiConstants.ANNOTATION_SINK,
                                sinkType, sinkAnnotation, SiddhiConstants.NAMESPACE_SINK);
                        ConfigReader sinkConfigReader = siddhiAppContext.getSiddhiContext().getConfigManager()
                                .generateConfigReader(sinkExtension.getNamespace(), sinkExtension.getName());
                        final boolean isDistributedTransport = (distributionAnnotation != null);
                        boolean isMultiClient = false;
                        if (isDistributedTransport) {
                            Sink sink = createSink(sinkExtension, siddhiAppContext);
                            isMultiClient = isMultiClientDistributedTransport(sink, streamDefinition,
                                    distributionAnnotation, siddhiAppContext);
                            supportedDynamicOptions = sink.getSupportedDynamicOptions();
                            destinationOptHolders = createDestinationOptionHolders(distributionAnnotation,
                                    streamDefinition, sink, siddhiAppContext);
                            destinationDeploymentProperties = createDestinationDeploymentProperties(
                                    distributionAnnotation, sink);
                        }

                        final String mapType = mapAnnotation.getElement(SiddhiConstants.ANNOTATION_ELEMENT_TYPE);
                        if (mapType != null) {
                            Sink sink;
                            if (isDistributedTransport) {
                                sink = (isMultiClient) ? new MultiClientDistributedSink() :
                                        new SingleClientDistributedSink();
                            } else {
                                sink = createSink(sinkExtension, siddhiAppContext);
                            }
                            if (supportedDynamicOptions == null) {
                                supportedDynamicOptions = sink.getSupportedDynamicOptions();
                            }

                            //load output mapper extension
                            Extension mapperExtension = constructExtension(streamDefinition,
                                    SiddhiConstants.ANNOTATION_MAP, mapType, sinkAnnotation,
                                    SiddhiConstants.NAMESPACE_SINK_MAPPER);
                            ConfigReader mapperConfigReader = siddhiAppContext.getSiddhiContext().getConfigManager()
                                    .generateConfigReader(sinkExtension.getNamespace(), sinkExtension.getName());

                            SinkMapper sinkMapper = (SinkMapper) SiddhiClassLoader.loadExtensionImplementation(
                                    mapperExtension, SinkMapperExecutorExtensionHolder.getInstance(siddhiAppContext));

                            io.siddhi.annotation.Extension sinkExt
                                    = sink.getClass().getAnnotation(io.siddhi.annotation.Extension.class);

                            OptionHolder transportOptionHolder = constructOptionHolder(streamDefinition, sinkAnnotation,
                                    sinkExt, supportedDynamicOptions, true);
                            Map<String, String> deploymentProperties = createDeploymentProperties(sinkAnnotation,
                                    sinkExt);
                            OptionHolder mapOptionHolder = constructOptionHolder(streamDefinition, mapAnnotation,
                                    sinkMapper.getClass().getAnnotation(io.siddhi.annotation.Extension.class),
                                    sinkMapper.getSupportedDynamicOptions(), false);
                            List<Element> payloadElementList = getPayload(mapAnnotation);

                            OptionHolder distributionOptHolder = null;
                            SinkHandlerManager sinkHandlerManager = siddhiAppContext.getSiddhiContext().
                                    getSinkHandlerManager();
                            SinkHandler sinkHandler = null;
                            if (sinkHandlerManager != null) {
                                sinkHandler = sinkHandlerManager.generateSinkHandler();
                            }

                            if (isDistributedTransport) {
                                distributionOptHolder = constructOptionHolder(streamDefinition, distributionAnnotation,
                                        sinkExt, supportedDynamicOptions, true);
                                String strategyType = distributionOptHolder
                                        .validateAndGetStaticValue(SiddhiConstants.DISTRIBUTION_STRATEGY_KEY);
                                Extension strategyExtension = constructExtension(streamDefinition,
                                        SiddhiConstants.ANNOTATION_SINK, strategyType, sinkAnnotation,
                                        SiddhiConstants.NAMESPACE_DISTRIBUTION_STRATEGY);
                                ConfigReader configReader = siddhiAppContext.getSiddhiContext().getConfigManager()
                                        .generateConfigReader(strategyExtension.getNamespace(),
                                                strategyExtension.getName());
                                DistributionStrategy distributionStrategy = (DistributionStrategy) SiddhiClassLoader
                                        .loadExtensionImplementation(
                                                strategyExtension, DistributionStrategyExtensionHolder.getInstance
                                                        (siddhiAppContext));
                                distributionStrategy.init(streamDefinition, transportOptionHolder,
                                        distributionOptHolder, destinationOptHolders, configReader);

                                ((DistributedTransport) sink).init(streamDefinition, sinkType,
                                        transportOptionHolder, sinkConfigReader, sinkMapper,
                                        mapType, mapOptionHolder, sinkHandler, payloadElementList,
                                        mapperConfigReader, streamJunction, siddhiAppContext, destinationOptHolders,
                                        sinkAnnotation, distributionStrategy, supportedDynamicOptions,
                                        deploymentProperties, destinationDeploymentProperties);
                            } else {
                                sink.init(streamDefinition, sinkType, transportOptionHolder, sinkConfigReader,
                                        sinkMapper, mapType, mapOptionHolder, sinkHandler, payloadElementList,
                                        mapperConfigReader, deploymentProperties, streamJunction, siddhiAppContext);

                            }

                            if (sinkHandlerManager != null) {
                                sinkHandlerManager.registerSinkHandler(sinkHandler.getId(), sinkHandler);
                            }

                            validateSinkMapperCompatibility(streamDefinition, sinkType, mapType, sink, sinkMapper,
                                    sinkAnnotation);

                            // Setting the output group determiner
                            OutputGroupDeterminer groupDeterminer = constructOutputGroupDeterminer(
                                    transportOptionHolder, distributionOptHolder, streamDefinition,
                                    destinationOptHolders.size());
                            if (groupDeterminer != null) {
                                sink.getMapper().setGroupDeterminer(groupDeterminer);
                            }

                            List<Sink> eventSinks = eventSinkMap.get(streamDefinition.getId());
                            if (eventSinks == null) {
                                eventSinks = new ArrayList<>();
                                eventSinks.add(sink);
                                eventSinkMap.put(streamDefinition.getId(), eventSinks);
                            } else {
                                eventSinks.add(sink);
                            }
                        } else {
                            throw new SiddhiAppCreationException(
                                    "Attribute 'type' does not exist for annotation '" + mapAnnotation + "'",
                                    mapAnnotation, siddhiAppContext);
                        }
                    } else {
                        throw new SiddhiAppCreationException("Both @sink(type=) and @map(type=) are required.",
                                sinkAnnotation.getQueryContextStartIndex(), sinkAnnotation.getQueryContextEndIndex());
                    }
                } catch (Throwable t) {
                    ExceptionUtil.populateQueryContext(t, sinkAnnotation, siddhiAppContext);
                    throw t;
                }
            }
        }
    }
Utility class for queryParser to help with QueryRuntime
generation.
modules/siddhi-core/src/main/java/io/siddhi/core/util/parser/helper/DefinitionParserHelper.java