Repository: apache/flink
File: flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/over/NonTimeRangeUnboundedPrecedingFunctionTest.java (Java)

Code before:
        testHarness.processElement(insertRecord("key2", 1L, 100L));
        testHarness.processElement(insertRecord("key2", 2L, 200L));

        // out of order record should trigger updates for all records after its inserted position
        testHarness.processElement(insertRecord("key1", 4L, 400L));

        List<RowData> expectedRows =
                Arrays.asList(
                        outputRecord(RowKind.INSERT, "key1", 1L, 100L, 1L),
                        outputRecord(RowKind.INSERT, "key1", 2L, 200L, 3L),
                        outputRecord(RowKind.INSERT, "key1", 5L, 500L, 8L),
                        outputRecord(RowKind.INSERT, "key1", 6L, 600L, 14L),
                        outputRecord(RowKind.INSERT, "key2", 1L, 100L, 1L),
                        outputRecord(RowKind.INSERT, "key2", 2L, 200L, 3L),
                        outputRecord(RowKind.INSERT, "key1", 4L, 400L, 7L),
                        outputRecord(RowKind.UPDATE_BEFORE, "key1", 5L, 500L, 8L),
                        outputRecord(RowKind.UPDATE_AFTER, "key1", 5L, 500L, 12L),
                        outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L, 14L),
                        outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L, 18L));

        List<RowData> actualRows = testHarness.extractOutputValues();

        validateRows(actualRows, expectedRows);
    }

    @Test
    public void testInsertOnlyRecordsWithCustomSortKeyAndLongSumAgg() throws Exception {
        NonTimeRangeUnboundedPrecedingFunction<RowData> function =
                new NonTimeRangeUnboundedPrecedingFunction<RowData>(
                        0,
                        aggsSumLongHandleFunction,
                        GENERATED_ROW_VALUE_EQUALISER,
                        GENERATED_SORT_KEY_EQUALISER,
                        GENERATED_SORT_KEY_COMPARATOR_ASC,
                        accTypes,
                        inputFieldTypes,
                        SORT_KEY_TYPES,
                        SORT_KEY_SELECTOR) {};
        KeyedProcessOperator<RowData, RowData, RowData> operator =
                new KeyedProcessOperator<>(function);

        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
                createTestHarness(operator);

        testHarness.open();

        // put some records
        testHarness.processElement(insertRecord("key1", 1L, 100L));
        testHarness.processElement(insertRecord("key1", 2L, 200L));
        testHarness.processElement(insertRecord("key1", 5L, 500L));
        testHarness.processElement(insertRecord("key1", 6L, 600L));

Review comment (suggestion):
```suggestion
    void testInsertOnlyRecordsWithCustomSortKeyAndLongSumAgg() throws Exception {
```
in fact there is no need for having `public` in case of junit5

Code after:
        testHarness.processElement(insertRecord("key2", 2L, 200L));

        // out of order record should trigger updates for all records after its inserted position
        testHarness.processElement(insertRecord("key1", 4L, 400L));

        List<RowData> expectedRows =
                List.of(
                        outputRecord(RowKind.INSERT, "key1", 1L, 100L, 1L),
                        outputRecord(RowKind.INSERT, "key1", 2L, 200L, 3L),
                        outputRecord(RowKind.INSERT, "key1", 5L, 500L, 8L),
                        outputRecord(RowKind.INSERT, "key1", 6L, 600L, 14L),
                        outputRecord(RowKind.INSERT, "key2", 1L, 100L, 1L),
                        outputRecord(RowKind.INSERT, "key2", 2L, 200L, 3L),
                        outputRecord(RowKind.INSERT, "key1", 4L, 400L, 7L),
                        outputRecord(RowKind.UPDATE_BEFORE, "key1", 5L, 500L, 8L),
                        outputRecord(RowKind.UPDATE_AFTER, "key1", 5L, 500L, 12L),
                        outputRecord(RowKind.UPDATE_BEFORE, "key1", 6L, 600L, 14L),
                        outputRecord(RowKind.UPDATE_AFTER, "key1", 6L, 600L, 18L));

        List<RowData> actualRows = testHarness.extractOutputValues();

        validateRows(actualRows, expectedRows);
    }

    @Test
    public void testInsertOnlyRecordsWithCustomSortKeyAndLongSumAgg() throws Exception {
        NonTimeRangeUnboundedPrecedingFunction<RowData> function =
                new NonTimeRangeUnboundedPrecedingFunction<RowData>(
                        0,
                        aggsSumLongHandleFunction,
                        GENERATED_ROW_VALUE_EQUALISER,
                        GENERATED_SORT_KEY_EQUALISER,
                        GENERATED_SORT_KEY_COMPARATOR_ASC,
                        accTypes,
                        inputFieldTypes,
                        SORT_KEY_TYPES,
                        SORT_KEY_SELECTOR) {};
        KeyedProcessOperator<RowData, RowData, RowData> operator =
                new KeyedProcessOperator<>(function);

        OneInputStreamOperatorTestHarness<RowData, RowData> testHarness =
                createTestHarness(operator);

        testHarness.open();

        // put some records
        testHarness.processElement(insertRecord("key1", 1L, 100L));
        testHarness.processElement(insertRecord("key1", 2L, 200L));
        testHarness.processElement(insertRecord("key1", 5L, 500L));
        testHarness.processElement(insertRecord("key1", 6L, 600L));
        testHarness.processElement(insertRecord("key2", 1L, 100L));