Repository: apache/flink
File: flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/operators/join/deltajoin/StreamingDeltaJoinOperatorTest.java (Java)

Code before:

        MyAsyncFunction.block();

        StreamRecord<RowData> leftRecord3 = insertRecord(700, true, "jklk1");
        testHarness.processElement1(leftRecord3);

        testHarness.snapshot(1L, 0L);
        assertThat(testHarness.numKeyedStateEntries()).isEqualTo(1);

        MyAsyncFunction.release();
        waitAllDataProcessed();

        testHarness.snapshot(2L, 0L);
        assertThat(testHarness.numKeyedStateEntries()).isEqualTo(0);

        final ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
        expectedOutput.add(insertRecord(100, true, "jklk1", "jklk1", 300, true));
        expectedOutput.add(insertRecord(200, true, "jklk1", "jklk1", 300, true));
        expectedOutput.add(insertRecord(700, true, "jklk1", "jklk1", 300, true));

        assertor.assertOutputEqualsSorted(
                "result mismatch", expectedOutput, testHarness.getOutput());
    }

    @Test
    void testMeetExceptionWhenLookup() throws Exception {
        Throwable expectedException = new IllegalStateException("Mock to fail");
        MyAsyncFunction.throwException(expectedException);

        StreamRecord<RowData> record = insertRecord(100, true, "jklk1");
        testHarness.processElement1(record);

        // IllegalStateException(Failed to wait all data processed)
        //  +- Exception(Could not complete the stream element ...)
        //    +- RuntimeException(Failed to lookup table)
        //      +- Actual Exception
        assertThatThrownBy(this::waitAllDataProcessed)
                .cause()
                .cause()
                .cause()
                .isEqualTo(expectedException);
    }

    private void waitAllDataProcessed() throws Exception {
        testHarness.endAllInputs();
        if (latestException.isPresent()) {
            throw new IllegalStateException(
                    "Failed to wait all data processed", latestException.get());
        }
    }


Review comment (suggestion):
We can also enrich the situations where errors occur, such as multiple insert records and throwing exceptions midway

Code after:

        MyAsyncFunction.block();

        StreamRecord<RowData> leftRecord3 = insertRecord(700, true, "jklk1");
        testHarness.processElement1(leftRecord3);

        testHarness.snapshot(1L, 0L);
        assertThat(testHarness.numKeyedStateEntries()).isEqualTo(1);

        MyAsyncFunction.release();
        waitAllDataProcessed();

        testHarness.snapshot(2L, 0L);
        assertThat(testHarness.numKeyedStateEntries()).isEqualTo(0);

        final ConcurrentLinkedQueue<Object> expectedOutput = new ConcurrentLinkedQueue<>();
        expectedOutput.add(insertRecord(100, true, "jklk1", "jklk1", 300, true));
        expectedOutput.add(insertRecord(200, true, "jklk1", "jklk1", 300, true));
        expectedOutput.add(insertRecord(700, true, "jklk1", "jklk1", 300, true));

        assertor.assertOutputEqualsSorted(
                "result mismatch", expectedOutput, testHarness.getOutput());
    }

    @Test
    void testMeetExceptionWhenLookup() throws Exception {
        Throwable expectedException = new IllegalStateException("Mock to fail");
        MyAsyncFunction.setExpectedThrownException(expectedException);

        StreamRecord<RowData> record = insertRecord(100, true, "jklk1");
        testHarness.processElement1(record);

        // IllegalStateException(Failed to wait all data processed)
        //  +- Exception(Could not complete the stream element ...)
        //    +- RuntimeException(Failed to lookup table)
        //      +- Actual Exception
        assertThatThrownBy(this::waitAllDataProcessed)
                .cause()
                .cause()
                .cause()
                .isEqualTo(expectedException);
    }

    private void waitAllDataProcessed() throws Exception {
        testHarness.endAllInputs();
        if (latestException.isPresent()) {
            throw new IllegalStateException(
                    "Failed to wait all data processed", latestException.get());
        }
    }