public ComplexEventChunk<StreamEvent> aggregateData(ComplexEventChunk<StreamEvent> retrievedData) {

        Set<String> groupByKeys = new HashSet<>();
        while (retrievedData.hasNext()) {
            StreamEvent streamEvent = retrievedData.next();
            String groupByKey = groupByKeyGenerator.constructEventKey(streamEvent);
            groupByKeys.add(groupByKey);
            SiddhiAppContext.startGroupByFlow(groupByKey);
            synchronized (this) {
                ValueState state = (ValueState) valueStateHolder.getState();
                try {
                    boolean shouldUpdate = true;
                    if (shouldUpdateTimestamp != null) {
                        shouldUpdate = shouldUpdate(shouldUpdateTimestamp.execute(streamEvent), state);
                    }
                    for (int i = 0; i < baseExecutors.size(); i++) { // keeping timestamp value location as null
                        ExpressionExecutor expressionExecutor = baseExecutors.get(i);
                        if (shouldUpdate) {
                            state.setValue(expressionExecutor.execute(streamEvent), i + 1);
                        } else if (!(expressionExecutor instanceof VariableExpressionExecutor)) {
                            state.setValue(expressionExecutor.execute(streamEvent), i + 1);
                        }
                    }
                } finally {
                    valueStateHolder.returnState(state);
                    SiddhiAppContext.stopGroupByFlow();
                }
            }
        }

        //clean all executors
        for (String groupByKey : groupByKeys) {
            SiddhiAppContext.startGroupByFlow(groupByKey);
            try {
                for (ExpressionExecutor expressionExecutor : baseExecutors) {
                    expressionExecutor.execute(resetEvent);
                }
            } finally {
                SiddhiAppContext.stopGroupByFlow();
            }
        }
        return createEventChunkFromAggregatedData();
    }
This class implements logic to process aggregates(after retrieval from tables) that were aggregated
using external timestamp.
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/OutOfOrderEventsDataAggregator.java