@Override
    public boolean hasNext() {
        return supplyWithLock(
                () -> {
                    if (cacheQueue.size() > 0) {
                        return true;
                    } else if (closed) {
                        return false;
                    } else {
                        waitCacheNotEmpty();
                        return hasNext();
                    }
                });
    }
@Test
    void testHasNext() throws ExecutionException, InterruptedException {
        CompletableFuture<Object> udfTrigger = new CompletableFuture<>();
        CompletableFuture<Object> udfReadIteratorFinishIdentifier = new CompletableFuture<>();
        CompletableFuture<Object> udfFinishTrigger = new CompletableFuture<>();
        MapPartitionIterator<String> iterator =
                new MapPartitionIterator<>(
                        inputIterator -> {
                            try {
                                udfTrigger.get();
                            } catch (InterruptedException | ExecutionException e) {
                                ExceptionUtils.rethrow(e);
                            }
                            for (int index = 0; index < RECORD_NUMBER; ++index) {
                                inputIterator.next();
                            }
                            udfReadIteratorFinishIdentifier.complete(null);
                            try {
                                udfFinishTrigger.get();
                            } catch (InterruptedException | ExecutionException e) {
                                ExceptionUtils.rethrow(e);
                            }
                        });
        // 1.Test hasNext() when the cache is not empty in the MapPartitionIterator.
        addRecordToIterator(RECORD_NUMBER, iterator);
        assertThat(iterator.hasNext()).isTrue();
        // 2.Test hasNext() when the cache is empty in the MapPartitionIterator.
        udfTrigger.complete(null);
        udfReadIteratorFinishIdentifier.get();
        assertThat(udfReadIteratorFinishIdentifier).isCompleted();
        CompletableFuture<Object> mockedUDFThread1 = new CompletableFuture<>();
        CompletableFuture<Boolean> hasNextFinishIdentifier1 = new CompletableFuture<>();
        mockedUDFThread1.thenRunAsync(
                () -> {
                    boolean hasNext = iterator.hasNext();
                    hasNextFinishIdentifier1.complete(hasNext);
                });
        mockedUDFThread1.complete(null);
        assertThat(hasNextFinishIdentifier1).isNotCompleted();
        iterator.addRecord(RECORD);
        hasNextFinishIdentifier1.get();
        assertThat(hasNextFinishIdentifier1).isCompletedWithValue(true);
        iterator.next();
        // 2.Test hasNext() when the MapPartitionIterator is closed.
        CompletableFuture<Object> mockedUDFThread2 = new CompletableFuture<>();
        CompletableFuture<Boolean> hasNextFinishIdentifier2 = new CompletableFuture<>();
        mockedUDFThread2.thenRunAsync(
                () -> {
                    boolean hasNext = iterator.hasNext();
                    hasNextFinishIdentifier2.complete(hasNext);
                    udfFinishTrigger.complete(null);
                });
        mockedUDFThread2.complete(null);
        assertThat(hasNextFinishIdentifier2).isNotCompleted();
        iterator.close();
        assertThat(hasNextFinishIdentifier2).isCompletedWithValue(false);
        assertThat(udfFinishTrigger).isCompleted();
    }