@Override
    protected void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor,
                           StreamEventCloner streamEventCloner, WindowState state) {
        List<ComplexEventChunk<StreamEvent>> streamEventChunks = new ArrayList<ComplexEventChunk<StreamEvent>>();
        ComplexEventChunk<StreamEvent> currentEventChunk = new ComplexEventChunk<StreamEvent>();
        synchronized (state) {
            long currentTime = siddhiQueryContext.getSiddhiAppContext().getTimestampGenerator().currentTime();
            if (outputExpectsExpiredEvents) {
                if (state.expiredEventQueue.getFirst() != null) {
                    while (state.expiredEventQueue.hasNext()) {
                        state.expiredEventQueue.next().setTimestamp(currentTime);
                    }
                    currentEventChunk.add(state.expiredEventQueue.getFirst());
                    if (state.resetEvent != null) {
                        currentEventChunk.add(state.resetEvent);
                        state.resetEvent = null;
                    }
                }
                state.expiredEventQueue.clear();
            }
            //check whether the streamEventChunk has next event before add into output stream event chunk
            ComplexEventChunk<StreamEvent> currentEventQueue = new ComplexEventChunk<>();
            int count = 0;
            if (streamEventChunk.hasNext()) {
                do {
                    StreamEvent streamEvent = streamEventChunk.next();
                    StreamEvent clonedStreamEventToProcess = streamEventCloner.copyStreamEvent(streamEvent);
                    if (outputExpectsExpiredEvents) {
                        StreamEvent clonedStreamEventToExpire = streamEventCloner.copyStreamEvent(streamEvent);
                        clonedStreamEventToExpire.setType(StreamEvent.Type.EXPIRED);
                        state.expiredEventQueue.add(clonedStreamEventToExpire);
                    }
                    currentEventQueue.add(clonedStreamEventToProcess);
                    count++;
                    if (count == length) {
                        if (currentEventQueue.getFirst() != null) {
                            if (state.resetEvent != null) {
                                currentEventChunk.add(state.resetEvent);
                            }
                            state.resetEvent = streamEventCloner.copyStreamEvent(currentEventQueue.getFirst());
                            state.resetEvent.setType(ComplexEvent.Type.RESET);
                            currentEventChunk.add(currentEventQueue.getFirst());
                        }
                        count = 0;
                        currentEventQueue.clear();
                        if (currentEventChunk.getFirst() != null) {
                            streamEventChunks.add(currentEventChunk);
                            currentEventChunk = new ComplexEventChunk<StreamEvent>();
                        }
                    }
                } while (streamEventChunk.hasNext());
                if (currentEventQueue.getFirst() != null) {
                    if (state.resetEvent != null) {
                        currentEventChunk.add(state.resetEvent);
                    }
                    state.resetEvent = streamEventCloner.copyStreamEvent(currentEventQueue.getFirst());
                    state.resetEvent.setType(ComplexEvent.Type.RESET);
                    currentEventChunk.add(currentEventQueue.getFirst());
                }
                if (currentEventChunk.getFirst() != null) {
                    streamEventChunks.add(currentEventChunk);
                }
            }
        }
        for (ComplexEventChunk<StreamEvent> processStreamEventChunk : streamEventChunks) {
            nextProcessor.process(processStreamEventChunk);
        }
    }
Implementation of {@link WindowProcessor} which represent a Batch Window that aggregate batch of incoming events
together.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/BatchWindowProcessor.java