@Override
    public void open(ExecutionContext ctx) throws Exception {
        super.open(ctx);
        equaliser =
                genRecordEqualiser.newInstance(ctx.getRuntimeContext().getUserCodeClassLoader());
    }
@Test
    public void testWithGenerateUpdateBeforeAndStateTtl() throws Exception {
        ProcTimeMiniBatchDeduplicateKeepLastRowFunction func =
                createFunction(true, true, minTime.toMilliseconds());
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(func);
        testHarness.setup();
        testHarness.open();

        testHarness.processElement(insertRecord("book", 1L, 10));
        testHarness.processElement(insertRecord("book", 2L, 11));
        // output is empty because bundle not trigger yet.
        assertThat(testHarness.getOutput()).isEmpty();
        testHarness.processElement(insertRecord("book", 1L, 13));

        testHarness.setStateTtlProcessingTime(30);
        testHarness.processElement(insertRecord("book", 1L, 17));
        testHarness.processElement(insertRecord("book", 2L, 18));
        testHarness.processElement(insertRecord("book", 1L, 19));

        List<Object> expectedOutput = new ArrayList<>();
        expectedOutput.add(insertRecord("book", 2L, 11));
        expectedOutput.add(insertRecord("book", 1L, 13));
        // because (2L,11), (1L,13) retired, so no UPDATE_BEFORE message send to downstream
        expectedOutput.add(insertRecord("book", 1L, 19));
        expectedOutput.add(insertRecord("book", 2L, 18));
        assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
    }