@Override
    public IN next() {
        return supplyWithLock(
                () -> {
                    IN record;
                    if (cacheQueue.size() > 0) {
                        if (!closed && cacheQueue.size() == DEFAULT_MAX_CACHE_NUM) {
                            cacheNotFull.signalAll();
                        }
                        record = cacheQueue.poll();
                        return record;
                    } else {
                        if (closed) {
                            return null;
                        }
                        waitCacheNotEmpty();
                        return cacheQueue.poll();
                    }
                });
    }
@Test
    void testNext() 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 next() when the cache is not 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 next() when the cache is empty in the MapPartitionIterator.
        CompletableFuture<Object> mockedUDFThread1 = new CompletableFuture<>();
        CompletableFuture<String> nextFinishIdentifier1 = new CompletableFuture<>();
        mockedUDFThread1.thenRunAsync(
                () -> {
                    String next = iterator.next();
                    nextFinishIdentifier1.complete(next);
                });
        mockedUDFThread1.complete(null);
        assertThat(nextFinishIdentifier1).isNotCompleted();
        iterator.addRecord(RECORD);
        nextFinishIdentifier1.get();
        assertThat(nextFinishIdentifier1).isCompletedWithValue(RECORD);
        // 2.Test next() when the MapPartitionIterator is closed.
        CompletableFuture<Object> mockedUDFThread2 = new CompletableFuture<>();
        CompletableFuture<String> nextFinishIdentifier2 = new CompletableFuture<>();
        mockedUDFThread2.thenRunAsync(
                () -> {
                    String next = iterator.next();
                    nextFinishIdentifier2.complete(next);
                    udfFinishTrigger.complete(null);
                });
        mockedUDFThread2.complete(null);
        assertThat(nextFinishIdentifier2).isNotCompleted();
        iterator.close();
        assertThat(nextFinishIdentifier2).isCompletedWithValue(null);
        assertThat(udfFinishTrigger).isCompleted();
    }