public void close() {
        runWithLock(
                () -> {
                    closed = true;
                    if (!udfFinished) {
                        cacheNotEmpty.signalAll();
                        waitUDFFinished();
                    }
                    udfExecutor.shutdown();
                });
    }
@Test
    void testClose() throws ExecutionException, InterruptedException {
        // 1.Test close() when the cache is not empty in the MapPartitionIterator.
        CompletableFuture<?> udfFinishTrigger1 = new CompletableFuture<>();
        MapPartitionIterator<String> iterator1 =
                new MapPartitionIterator<>(
                        ignored -> {
                            try {
                                udfFinishTrigger1.get();
                            } catch (InterruptedException | ExecutionException e) {
                                ExceptionUtils.rethrow(e);
                            }
                        });
        iterator1.addRecord(RECORD);
        CompletableFuture<Object> mockedTaskThread1 = new CompletableFuture<>();
        CompletableFuture<Object> iteratorCloseIdentifier1 = new CompletableFuture<>();
        mockedTaskThread1.thenRunAsync(
                () -> {
                    iterator1.close();
                    iteratorCloseIdentifier1.complete(null);
                });
        mockedTaskThread1.complete(null);
        assertThat(iteratorCloseIdentifier1).isNotCompleted();
        udfFinishTrigger1.complete(null);
        iteratorCloseIdentifier1.get();
        assertThat(iteratorCloseIdentifier1).isCompleted();
        // 2.Test close() when the cache is empty in the MapPartitionIterator.
        CompletableFuture<?> udfFinishTrigger2 = new CompletableFuture<>();
        MapPartitionIterator<String> iterator2 =
                new MapPartitionIterator<>(
                        ignored -> {
                            try {
                                udfFinishTrigger2.get();
                            } catch (InterruptedException | ExecutionException e) {
                                ExceptionUtils.rethrow(e);
                            }
                        });
        CompletableFuture<Object> mockedTaskThread2 = new CompletableFuture<>();
        CompletableFuture<Object> iteratorCloseIdentifier2 = new CompletableFuture<>();
        mockedTaskThread1.thenRunAsync(
                () -> {
                    iterator2.close();
                    iteratorCloseIdentifier2.complete(null);
                });
        mockedTaskThread2.complete(null);
        assertThat(iteratorCloseIdentifier2).isNotCompleted();
        udfFinishTrigger2.complete(null);
        iteratorCloseIdentifier2.get();
        assertThat(iteratorCloseIdentifier2).isCompleted();
        // 2.Test close() when the udf is finished in the MapPartitionIterator.
        MapPartitionIterator<String> iterator3 = new MapPartitionIterator<>(ignored -> {});
        iterator3.close();
    }