@Override
    public void resetState() {
        StreamPreState state = stateHolder.getState();
        lock.lock();
        try {
            state.pendingStateEventList.clear();
            if (isStartState && state.newAndEveryStateEventList.isEmpty()) {
                if (stateType == StateInputStream.Type.SEQUENCE && thisStatePostProcessor.nextEveryStatePreProcessor ==
                        null && !((StreamPreStateProcessor) thisStatePostProcessor.nextStatePreProcessor).
                        getPendingStateEventList().isEmpty()) {
                    return;
                }
                init();
            }
        } finally {
            lock.unlock();
            stateHolder.returnState(state);
        }
    }
Set as the last element of the processor chain

@param processor Last processor in the chain
modules/siddhi-core/src/main/java/io/siddhi/core/query/input/stream/state/StreamPreStateProcessor.java