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:
        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());
        }
    }

    private KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData, RowData>
            createDeltaJoinOperatorTestHarness() throws Exception {
        TaskMailbox mailbox = new TaskMailboxImpl();
        MailboxProcessor mailboxProcessor =
                new MailboxProcessor(controller -> {}, mailbox, StreamTaskActionExecutor.IMMEDIATE);

        DataStructureConverter<RowData, Object> leftFetcherConverter =
                (DataStructureConverter)
                        DataStructureConverters.getConverter(leftTypeInfo.getDataType());

        AsyncDeltaJoinRunner leftAsyncFunction =
                new AsyncDeltaJoinRunner(
                        new GeneratedFunctionWrapper<>(new MyAsyncFunction()),
                        leftFetcherConverter,
                        new GeneratedResultFutureWrapper<>(new TestingFetcherResultFuture()),
                        leftTypeInfo.toRowSerializer(),
                        AEC_CAPACITY,
                        false);

        DataStructureConverter<RowData, Object> rightFetcherConverter =

Review comment (bug):
Correspondingly, we can use getActualExternalFailureCause().isPresent() to judge if we should throw exception here.

Code after:
        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());
        }
    }

    private KeyedTwoInputStreamOperatorTestHarness<RowData, RowData, RowData, RowData>
            createDeltaJoinOperatorTestHarness() throws Exception {
        TaskMailbox mailbox = new TaskMailboxImpl();
        MailboxProcessor mailboxProcessor =
                new MailboxProcessor(controller -> {}, mailbox, StreamTaskActionExecutor.IMMEDIATE);

        DataStructureConverter<RowData, Object> leftFetcherConverter =
                (DataStructureConverter)
                        DataStructureConverters.getConverter(leftTypeInfo.getDataType());

        AsyncDeltaJoinRunner leftAsyncFunction =
                new AsyncDeltaJoinRunner(
                        new GeneratedFunctionWrapper<>(new MyAsyncFunction()),
                        leftFetcherConverter,
                        new GeneratedResultFutureWrapper<>(new TestingFetcherResultFuture()),
                        leftTypeInfo.toRowSerializer(),
                        AEC_CAPACITY,
                        false);

        DataStructureConverter<RowData, Object> rightFetcherConverter =