@Override
    public void open() throws Exception {
        super.open();
        windowSerializer = windowAssigner.getWindowSerializer(new ExecutionConfig());

        internalTimerService = getInternalTimerService("window-timers", windowSerializer, this);

        triggerContext = new TriggerContext();
        triggerContext.open();

        StateDescriptor<ListState<RowData>, List<RowData>> windowStateDescriptor =
                new ListStateDescriptor<>("window-input", new RowDataSerializer(inputType));
        StateDescriptor<ListState<RowData>, List<RowData>> dataRetractStateDescriptor =
                new ListStateDescriptor<>("data-retract", new RowDataSerializer(inputType));
        this.windowAccumulateData =
                (InternalListState<K, W, RowData>)
                        getOrCreateKeyedState(windowSerializer, windowStateDescriptor);
        this.windowRetractData =
                (InternalListState<K, W, RowData>)
                        getOrCreateKeyedState(windowSerializer, dataRetractStateDescriptor);
        inputKeyAndWindow = new LinkedList<>();
        windowProperty = new GenericRowData(namedProperties.length);
        windowAggResult = new JoinedRowData();

        WindowContext windowContext = new WindowContext();
        windowAssigner.open(windowContext);
    }
@Test
    void testGroupWindowAggregateFunction() throws Exception {
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
                getTestHarness(new Configuration());
        long initialTime = 0L;
        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
        testHarness.open();
        testHarness.processElement(
                new StreamRecord<>(newBinaryRow(true, "c1", "c2", 0L, 0L), initialTime + 1));
        testHarness.processElement(
                new StreamRecord<>(newBinaryRow(true, "c1", "c4", 1L, 6000L), initialTime + 2));
        testHarness.processElement(
                new StreamRecord<>(newBinaryRow(true, "c1", "c6", 2L, 10000L), initialTime + 3));
        testHarness.processElement(
                new StreamRecord<>(newBinaryRow(true, "c2", "c8", 3L, 0L), initialTime + 4));
        testHarness.processElement(
                new StreamRecord<>(newBinaryRow(true, "c3", "c8", 3L, 0L), initialTime + 5));
        testHarness.processElement(
                new StreamRecord<>(newBinaryRow(false, "c3", "c8", 3L, 0L), initialTime + 6));
        testHarness.processWatermark(Long.MAX_VALUE);
        testHarness.close();

        expectedOutput.add(
                new StreamRecord<>(
                        newRow(
                                true,
                                "c1",
                                0L,
                                TimestampData.fromEpochMillis(-5000L),
                                TimestampData.fromEpochMillis(5000L))));

        expectedOutput.add(
                new StreamRecord<>(
                        newRow(
                                true,
                                "c2",
                                3L,
                                TimestampData.fromEpochMillis(-5000L),
                                TimestampData.fromEpochMillis(5000L))));

        expectedOutput.add(
                new StreamRecord<>(
                        newRow(
                                true,
                                "c1",
                                0L,
                                TimestampData.fromEpochMillis(0L),
                                TimestampData.fromEpochMillis(10000L))));

        expectedOutput.add(
                new StreamRecord<>(
                        newRow(
                                true,
                                "c2",
                                3L,
                                TimestampData.fromEpochMillis(0L),
                                TimestampData.fromEpochMillis(10000L))));

        expectedOutput.add(
                new StreamRecord<>(
                        newRow(
                                true,
                                "c1",
                                1L,
                                TimestampData.fromEpochMillis(5000L),
                                TimestampData.fromEpochMillis(15000L))));
        expectedOutput.add(
                new StreamRecord<>(
                        newRow(
                                true,
                                "c1",
                                2L,
                                TimestampData.fromEpochMillis(10000L),
                                TimestampData.fromEpochMillis(20000L))));

        expectedOutput.add(new Watermark(Long.MAX_VALUE));

        assertOutputEquals("Output was not correct.", expectedOutput, testHarness.getOutput());
    }