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:
        private static ExecutorService executorService;

        private static @Nullable CountDownLatch lock;

        private static final AtomicInteger leftInvokeCount = new AtomicInteger(0);

        private static final AtomicInteger rightInvokeCount = new AtomicInteger(0);

        private static Optional<Throwable> expectedThrownException = Optional.empty();

        // ===== runtime info =====
        private Boolean treatRightAsLookupTable;

        public void tagInvokingSideDuringRuntime(boolean isLeftInvoking) {
            this.treatRightAsLookupTable = isLeftInvoking;
        }

        public static void block() throws Exception {
            lock = new CountDownLatch(1);
        }

        public static void release() {
            Objects.requireNonNull(lock).countDown();
        }

        public static void throwException(Throwable t) {
            expectedThrownException = Optional.of(t);
        }

        public static void clearException() {
            expectedThrownException = Optional.empty();
        }

        @Override
        public void asyncInvoke(final RowData input, final ResultFuture<Object> resultFuture) {
            executorService.submit(
                    () -> {
                        try {
                            if (expectedThrownException.isPresent()) {
                                throw expectedThrownException.get();
                            }

                            if (lock != null) {
                                lock.await();
                            }

                            LinkedList<RowData> lookupTableData;
                            RowDataKeySelector streamSideJoinKeySelector;
                            RowDataKeySelector lookupSideJoinKeySelector;
                            if (Objects.requireNonNull(treatRightAsLookupTable)) {
                                synchronized (rightTableCurrentData) {

Review comment (suggestion):
nit: throwException implied it will throw an Exception. I suggest rename to `setExpectedThrownException `or storeException

Code after:
        private static ExecutorService executorService;

        private static @Nullable CountDownLatch lock;

        private static final AtomicInteger leftInvokeCount = new AtomicInteger(0);

        private static final AtomicInteger rightInvokeCount = new AtomicInteger(0);

        private static Optional<Throwable> expectedThrownException = Optional.empty();

        // ===== runtime info =====
        private Boolean treatRightAsLookupTable;

        public void tagInvokingSideDuringRuntime(boolean isLeftInvoking) {
            this.treatRightAsLookupTable = isLeftInvoking;
        }

        public static void block() throws Exception {
            lock = new CountDownLatch(1);
        }

        public static void release() {
            Objects.requireNonNull(lock).countDown();
        }

        public static void setExpectedThrownException(Throwable t) {
            expectedThrownException = Optional.of(t);
        }

        public static void clearExpectedThrownException() {
            expectedThrownException = Optional.empty();
        }

        @Override
        public void asyncInvoke(final RowData input, final ResultFuture<Object> resultFuture) {
            executorService.submit(
                    () -> {
                        try {
                            if (expectedThrownException.isPresent()) {
                                throw expectedThrownException.get();
                            }

                            if (lock != null) {
                                lock.await();
                            }

                            LinkedList<RowData> lookupTableData;
                            RowDataKeySelector streamSideJoinKeySelector;
                            RowDataKeySelector lookupSideJoinKeySelector;
                            if (Objects.requireNonNull(treatRightAsLookupTable)) {
                                synchronized (rightTableCurrentData) {