@Override
    public void open() throws Exception {
        windowSerializer = windowAssigner.getWindowSerializer(new ExecutionConfig());
        internalTimerService = getInternalTimerService("window-timers", windowSerializer, this);
        // The structure is:  [type]|[normal record]|[timestamp]|[current watermark]|[timer data]
        // If the type is 'NORMAL_RECORD', store the RowData object in the 2nd column.
        // If the type is 'TRIGGER_TIMER', store the timestamp in 3rd column and the timer
        // data in 5th column.
        reuseRowData =
                new UpdatableRowData(GenericRowData.of(NORMAL_RECORD, null, null, null, null), 5);
        reuseTimerRowData =
                new UpdatableRowData(GenericRowData.of(TRIGGER_TIMER, null, null, null, null), 5);
        // The structure is:  [timer_type]|[row key]|[encoded namespace]
        reuseTimerData = new UpdatableRowData(GenericRowData.of(0, null, 0), 3);
        reuseTimerRowData.setField(4, reuseTimerData);
        keyLength = getKeyType().getFieldCount();
        keySerializer = (RowDataSerializer) getKeySerializer();
        super.open();
    }
@Test
    void testFinishBundleTriggeredByTime() throws Exception {
        Configuration conf = new Configuration();
        conf.set(PythonOptions.MAX_BUNDLE_SIZE, 10);
        conf.set(PythonOptions.MAX_BUNDLE_TIME_MILLS, 1000L);
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = getTestHarness(conf);

        long initialTime = 0L;
        ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();

        testHarness.open();

        testHarness.processElement(newRecord(true, initialTime + 1, "c1", "c2", 0L, 0L));
        testHarness.processElement(newRecord(true, initialTime + 2, "c1", "c4", 1L, 6000L));
        testHarness.processElement(newRecord(true, initialTime + 3, "c1", "c6", 2L, 10000L));
        testHarness.processElement(newRecord(true, initialTime + 4, "c2", "c8", 3L, 0L));
        testHarness.processWatermark(new Watermark(20000L));

        testHarness.setProcessingTime(1000L);
        expectedOutput.add(newWindowRecord(-5000L, 5000L, "c1", 0L));
        expectedOutput.add(newStateCleanupRecord(-5000L, 5000L, "c1"));
        expectedOutput.add(newWindowRecord(-5000L, 5000L, "c2", 3L));
        expectedOutput.add(newStateCleanupRecord(-5000L, 5000L, "c2"));
        expectedOutput.add(newWindowRecord(0, 10000L, "c2", 3L));
        expectedOutput.add(newStateCleanupRecord(0L, 10000L, "c2"));
        expectedOutput.add(newWindowRecord(0, 10000L, "c1", 0L));
        expectedOutput.add(newStateCleanupRecord(0L, 10000L, "c1"));
        expectedOutput.add(newWindowRecord(5000L, 15000L, "c1", 1L));
        expectedOutput.add(newStateCleanupRecord(5000L, 15000L, "c1"));
        expectedOutput.add(newWindowRecord(10000L, 20000L, "c1", 2L));
        expectedOutput.add(newStateCleanupRecord(10000L, 20000L, "c1"));

        expectedOutput.add(new Watermark(20000L));

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

        testHarness.close();
    }