@Override
    public void processElement(RowData input, Context ctx, Collector<RowData> out)
            throws Exception {
        processFirstRowOnProcTime(input, state, out);
    }
@Test
    public void test() throws Exception {
        ProcTimeDeduplicateKeepFirstRowFunction func =
                new ProcTimeDeduplicateKeepFirstRowFunction(minTime.toMilliseconds());
        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness = createTestHarness(func);
        testHarness.open();
        testHarness.processElement(insertRecord("book", 1L, 12));
        testHarness.processElement(insertRecord("book", 2L, 11));
        testHarness.processElement(insertRecord("book", 1L, 13));
        testHarness.close();

        // Keep FirstRow in deduplicate will not send retraction
        List<Object> expectedOutput = new ArrayList<>();
        expectedOutput.add(insertRecord("book", 1L, 12));
        expectedOutput.add(insertRecord("book", 2L, 11));
        assertor.assertOutputEqualsSorted("output wrong.", expectedOutput, testHarness.getOutput());
    }