private synchronized ComplexEventChunk<StreamEvent> getProcessedEventChunk() {
        ComplexEventChunk<StreamEvent> streamEventChunk = new ComplexEventChunk<>();
        Map<String, State> valueStoreMap = this.valueStateHolder.getAllGroupByStates();
        try {
            for (State aState : valueStoreMap.values()) {
                ValueState state = (ValueState) aState;
                StreamEvent streamEvent = streamEventFactory.newInstance();
                long timestamp = state.lastTimestamp;
                streamEvent.setTimestamp(timestamp);
                state.setValue(timestamp, 0);
                streamEvent.setOutputData(state.values);
                streamEventChunk.add(streamEvent);
            }
        } finally {
            this.valueStateHolder.returnGroupByStates(valueStoreMap);
        }
        return streamEventChunk;
    }
This class implements the logic to aggregate data that is in-memory or in tables, in incremental data processing.
<p>
In-memory data is required to be aggregated when retrieving values from an aggregate.
Table data (tables used to persist incremental aggregates) needs to be aggregated when querying aggregate data from
a different server (apart from the server, which was used to define the aggregation).
modules/siddhi-core/src/main/java/io/siddhi/core/aggregation/IncrementalDataAggregator.java