public void addRecord(IN record) {
        runWithLock(
                () -> {
                    checkState(!closed);
                    if (udfFinished) {
                        return;
                    }
                    if (cacheQueue.size() < DEFAULT_MAX_CACHE_NUM) {
                        cacheQueue.add(record);
                        if (cacheQueue.size() == 1) {
                            cacheNotEmpty.signalAll();
                        }
                    } else {
                        waitCacheNotFull();
                        addRecord(record);
                    }
                });
    }
@Test
    void testAddRecord() throws ExecutionException, InterruptedException {
        CompletableFuture<List<String>> result = new CompletableFuture<>();
        CompletableFuture<Object> udfFinishTrigger = new CompletableFuture<>();
        MapPartitionIterator<String> iterator =
                new MapPartitionIterator<>(
                        inputIterator -> {
                            List<String> strings = new ArrayList<>();
                            for (int index = 0; index < RECORD_NUMBER; ++index) {
                                strings.add(inputIterator.next());
                            }
                            result.complete(strings);
                            try {
                                udfFinishTrigger.get();
                            } catch (InterruptedException | ExecutionException e) {
                                ExceptionUtils.rethrow(e);
                            }
                        });
        // 1.Test addRecord() when the cache is empty in the MapPartitionIterator.
        addRecordToIterator(RECORD_NUMBER, iterator);
        List<String> results = result.get();
        assertThat(results.size()).isEqualTo(RECORD_NUMBER);
        assertThat(results.get(0)).isEqualTo(RECORD);
        assertThat(results.get(1)).isEqualTo(RECORD);
        assertThat(results.get(2)).isEqualTo(RECORD);
        // 2.Test addRecord() when the cache is full in the MapPartitionIterator.
        addRecordToIterator(DEFAULT_MAX_CACHE_NUM, iterator);
        CompletableFuture<Object> mockedTaskThread1 = new CompletableFuture<>();
        CompletableFuture<List<String>> addRecordFinishIdentifier1 = new CompletableFuture<>();
        mockedTaskThread1.thenRunAsync(
                () -> {
                    iterator.addRecord(RECORD);
                    addRecordFinishIdentifier1.complete(null);
                });
        mockedTaskThread1.complete(null);
        assertThat(addRecordFinishIdentifier1).isNotCompleted();
        iterator.next();
        addRecordFinishIdentifier1.get();
        assertThat(addRecordFinishIdentifier1).isCompleted();
        // 2.Test addRecord() when the udf is finished in the MapPartitionIterator.
        CompletableFuture<Object> mockedTaskThread2 = new CompletableFuture<>();
        CompletableFuture<List<String>> addRecordFinishIdentifier2 = new CompletableFuture<>();
        mockedTaskThread2.thenRunAsync(
                () -> {
                    iterator.addRecord(RECORD);
                    addRecordFinishIdentifier2.complete(null);
                });
        mockedTaskThread2.complete(null);
        assertThat(addRecordFinishIdentifier2).isNotCompleted();
        udfFinishTrigger.complete(null);
        addRecordFinishIdentifier2.get();
        assertThat(addRecordFinishIdentifier2).isCompleted();
        assertThat(udfFinishTrigger).isCompleted();
        iterator.close();
    }