@Override
    public void process(ComplexEventChunk complexEventChunk) {
        ComplexEventChunk<StreamEvent> streamEventChunk =
                new ComplexEventChunk<>();
        try {
            int noOfEvents = 0;
            if (latencyTrackerInsert != null && Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                latencyTrackerInsert.markIn();
            }
            while (complexEventChunk.hasNext()) {
                ComplexEvent complexEvent = complexEventChunk.next();
                if (!isFirstEventArrived) {
                    aggregationRuntime.initialiseExecutors(true);
                    isFirstEventArrived = true;
                }
                StreamEvent newEvent = streamEventFactory.newInstance();
                for (int i = 0; i < incomingExpressionExecutors.size(); i++) {
                    ExpressionExecutor expressionExecutor = incomingExpressionExecutors.get(i);
                    Object outputData = expressionExecutor.execute(complexEvent);
                    if (expressionExecutor instanceof IncrementalUnixTimeFunctionExecutor && outputData == null) {
                        throw new SiddhiAppRuntimeException("Cannot retrieve the timestamp of event");
                    }
                    newEvent.setOutputData(outputData, i);
                }
                streamEventChunk.add(newEvent);
                noOfEvents++;
            }
            aggregationRuntime.processEvents(streamEventChunk);
            if (throughputTrackerInsert != null &&
                    Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                throughputTrackerInsert.eventsIn(noOfEvents);
            }
        } finally {
            if (latencyTrackerInsert != null && Level.BASIC.compareTo(siddhiAppContext.getRootMetricsLevel()) <= 0) {
                latencyTrackerInsert.markOut();
            }
        }

    }
Incremental Aggregation Processor to consume events to Incremental Aggregators.
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalAggregationProcessor.java