@Override
    protected StateFactory init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                                StreamEventClonerHolder streamEventClonerHolder,
                                boolean outputExpectsExpiredEvents, boolean findToBeExecuted,
                                SiddhiQueryContext siddhiQueryContext) {
        this.outputExpectsExpiredEvents = outputExpectsExpiredEvents;
        this.findToBeExecuted = findToBeExecuted;

        if (attributeExpressionExecutors.length >= 2 && attributeExpressionExecutors.length <= 5) {

            if (!(attributeExpressionExecutors[0] instanceof VariableExpressionExecutor)) {
                throw new SiddhiAppValidationException("ExternalTime window's 1st parameter timestamp should be a" +
                        " variable, but found " + attributeExpressionExecutors[0].getClass());
            }
            if (attributeExpressionExecutors[0].getReturnType() != Attribute.Type.LONG) {
                throw new SiddhiAppValidationException("ExternalTime window's 1st parameter timestamp should be " +
                        "type long, but found " + attributeExpressionExecutors[0].getReturnType());
            }
            timestampExpressionExecutor = (VariableExpressionExecutor) attributeExpressionExecutors[0];

            if (attributeExpressionExecutors[1].getReturnType() == Attribute.Type.INT) {
                timeToKeep = (Integer) ((ConstantExpressionExecutor) attributeExpressionExecutors[1]).getValue();

            } else if (attributeExpressionExecutors[1].getReturnType() == Attribute.Type.LONG) {
                timeToKeep = (Long) ((ConstantExpressionExecutor) attributeExpressionExecutors[1]).getValue();
            } else {
                throw new SiddhiAppValidationException("ExternalTimeBatch window's 2nd parameter windowTime " +
                        "should be either int or long, but found " + attributeExpressionExecutors[1].getReturnType());
            }

            if (attributeExpressionExecutors.length >= 3) {
                isStartTimeEnabled = true;
                if ((attributeExpressionExecutors[2] instanceof ConstantExpressionExecutor)) {
                    if (attributeExpressionExecutors[2].getReturnType() == Attribute.Type.INT) {
                        commonStartTime = Integer.parseInt(String.valueOf(((ConstantExpressionExecutor)
                                attributeExpressionExecutors[2]).getValue()));
                    } else if (attributeExpressionExecutors[2].getReturnType() == Attribute.Type.LONG) {
                        commonStartTime = Long.parseLong(String.valueOf(((ConstantExpressionExecutor)
                                attributeExpressionExecutors[2]).getValue()));
                    } else {
                        throw new SiddhiAppValidationException("ExternalTimeBatch window's 3rd parameter " +
                                "startTime should either be a constant (of type int or long) or an attribute (of type" +
                                " long), but found " + attributeExpressionExecutors[2].getReturnType());
                    }
                } else if (attributeExpressionExecutors[2].getReturnType() != Attribute.Type.LONG) {
                    throw new SiddhiAppValidationException("ExternalTimeBatch window's 3rd parameter startTime " +
                            "should either be a constant (of type int or long) or an attribute (of type long), but " +
                            "found " + attributeExpressionExecutors[2].getReturnType());
                } else {
                    startTimeAsVariable = attributeExpressionExecutors[2];
                }
            }

            if (attributeExpressionExecutors.length >= 4) {
                if (attributeExpressionExecutors[3].getReturnType() == Attribute.Type.INT) {
                    schedulerTimeout = Integer.parseInt(String.valueOf(((ConstantExpressionExecutor)
                            attributeExpressionExecutors[3]).getValue()));
                } else if (attributeExpressionExecutors[3].getReturnType() == Attribute.Type.LONG) {
                    schedulerTimeout = Long.parseLong(String.valueOf(((ConstantExpressionExecutor)
                            attributeExpressionExecutors[3]).getValue()));
                } else {
                    throw new SiddhiAppValidationException("ExternalTimeBatch window's 4th parameter timeout " +
                            "should be either int or long, but found " + attributeExpressionExecutors[3]
                            .getReturnType());
                }
            }

            if (attributeExpressionExecutors.length == 5) {
                if (attributeExpressionExecutors[4].getReturnType() == Attribute.Type.BOOL) {
                    replaceTimestampWithBatchEndTime = Boolean.parseBoolean(String.valueOf((
                            (ConstantExpressionExecutor) attributeExpressionExecutors[4]).getValue()));
                } else {
                    throw new SiddhiAppValidationException("ExternalTimeBatch window's 5th parameter " +
                            "replaceTimestampWithBatchEndTime should be bool, but found " +
                            attributeExpressionExecutors[4].getReturnType());
                }
            }
        } else {
            throw new SiddhiAppValidationException("ExternalTimeBatch window should only have two to five " +
                    "parameters (<long> timestamp, <int|long|time> windowTime, <long> startTime, <int|long|time> " +
                    "timeout, <bool> replaceTimestampWithBatchEndTime), but found " + attributeExpressionExecutors
                    .length + " input attributes");
        }
        return () -> new WindowState(outputExpectsExpiredEvents, schedulerTimeout, commonStartTime);
    }
Implementation of {@link WindowProcessor} which represent a Batch Window operating based on external time.
modules/siddhi-core/src/main/java/io/siddhi/core/query/processor/stream/window/ExternalTimeBatchWindowProcessor.java