private void dispatchEvent(long startTimeOfNewAggregates, long emittedTime, String timeZone) {
        ZonedDateTime startTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(startTimeOfNewAggregates),
                ZoneId.of(timeZone));
        ZonedDateTime endTime = ZonedDateTime.ofInstant(Instant.ofEpochMilli(emittedTime),
                ZoneId.of(timeZone));
        log.info("Aggregation event dispatched for the duration " + duration + " for aggregation " + aggregatorName +
                " to aggregate data from " + startTime + " to " + endTime + " ");
        ComplexEventChunk complexEventChunk = new ComplexEventChunk();
        StreamEvent streamEvent = streamEventFactory.newInstance();
        streamEvent.setType(ComplexEvent.Type.CURRENT);
        streamEvent.setTimestamp(emittedTime);
        List<Object> outputDataList = new ArrayList<>();
        outputDataList.add(startTimeOfNewAggregates);
        outputDataList.add(emittedTime);
        outputDataList.add(null);
        streamEvent.setOutputData(outputDataList.toArray());

        if (isProcessingExecutor) {
            complexEventChunk.add(streamEvent);
            cudStreamProcessorQueue.add(new QueuedCudStreamProcessor(cudStreamProcessor, streamEvent,
                    startTimeOfNewAggregates, emittedTime, timeZone, duration));
        }
        if (getNextExecutor() != null) {
            next.execute(complexEventChunk);
        }
    }
Incremental Executor implementation class for Persisted Aggregation
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/persistedaggregation/PersistedIncrementalExecutor.java