Repository: apache/flink
File: flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/aggregate/window/processors/WindowAggProcessorBase.java (Java)

Code before:
        this.shiftTimeZone = shiftTimeZone;
        this.useDayLightSaving = TimeZone.getTimeZone(shiftTimeZone).useDaylightTime();
    }

    @Override
    public void open(C context) throws Exception {
        this.ctx = context;
        this.clockService = ClockService.of(ctx.getTimerService());

        this.aggregator =
                genAggsHandler.newInstance(ctx.getRuntimeContext().getUserCodeClassLoader());

        this.reuseOutput = new JoinedRowData();
        this.currentProgress = Long.MIN_VALUE;
        this.windowTimerService = getWindowTimerService();

        prepareAggregator();
    }

    @Override
    public void initializeWatermark(long watermark) {
        if (isEventTime) {
            currentProgress = watermark;
            // Restore the watermark of timerService to prevent expired data from being treated as
            // not expired when flushWindowBuffer is executed.
            windowTimerService.initializeWatermark(watermark);
        }
    }

    @Override
    public void close() throws Exception {
        if (aggregator != null) {
            aggregator.close();
        }
    }

    protected abstract void prepareAggregator() throws Exception;

    protected abstract WindowTimerService<W> getWindowTimerService();

    /**
     * Send result to downstream.
     *
     * <p>The {@link org.apache.flink.types.RowKind} of the results is always {@link
     * org.apache.flink.types.RowKind#INSERT}.
     *
     * <p>TODO support early fire / late file to produce changelog result.
     */
    protected void collect(RowData currentKey, RowData aggResult) {
        reuseOutput.replace(currentKey, aggResult);
        ctx.output(reuseOutput);

Review comment (refactor):
nit: Can we move this logic to the part where the `internalTimerService` is created in `WindowAggOperator#open`? I think it might be easier to maintain.

Code after:
        this.shiftTimeZone = shiftTimeZone;
        this.useDayLightSaving = TimeZone.getTimeZone(shiftTimeZone).useDaylightTime();
    }

    @Override
    public void open(C context) throws Exception {
        this.ctx = context;
        this.clockService = ClockService.of(ctx.getTimerService());

        this.aggregator =
                genAggsHandler.newInstance(ctx.getRuntimeContext().getUserCodeClassLoader());

        this.reuseOutput = new JoinedRowData();
        this.currentProgress = Long.MIN_VALUE;
        this.windowTimerService = getWindowTimerService();

        prepareAggregator();
    }

    @Override
    public void initializeWatermark(long watermark) {
        if (isEventTime) {
            currentProgress = watermark;
        }
    }

    @Override
    public void close() throws Exception {
        if (aggregator != null) {
            aggregator.close();
        }
    }

    protected abstract void prepareAggregator() throws Exception;

    protected abstract WindowTimerService<W> getWindowTimerService();

    /**
     * Send result to downstream.
     *
     * <p>The {@link org.apache.flink.types.RowKind} of the results is always {@link
     * org.apache.flink.types.RowKind#INSERT}.
     *
     * <p>TODO support early fire / late file to produce changelog result.
     */
    protected void collect(RowData currentKey, RowData aggResult) {
        reuseOutput.replace(currentKey, aggResult);
        ctx.output(reuseOutput);
    }
}