public synchronized ResultSet fetchResults(FetchOrientation orientation, int maxFetchSize) {
        long token;
        switch (orientation) {
            case FETCH_NEXT:
                token = currentToken;
                break;
            case FETCH_PRIOR:
                token = currentToken - 1;
                break;
            default:
                throw new UnsupportedOperationException(
                        String.format("Unknown fetch orientation: %s.", orientation));
        }

        if (orientation == FetchOrientation.FETCH_NEXT && bufferedResults.isEmpty()) {
            // make sure data is available in the buffer
            resultStore.waitUntilHasData();
        }

        return fetchResults(token, maxFetchSize);
    }
@Test
    void testFetchResultInOrientationInParallel() throws Exception {
        List<Iterator<RowData>> dataSuppliers =
                data.stream()
                        .map(
                                row ->
                                        new TestIterator(
                                                () -> {
                                                    try {
                                                        Thread.sleep(1);
                                                        return row;
                                                    } catch (Exception e) {
                                                        throw new SqlExecutionException(
                                                                "Failed to return the row.", e);
                                                    }
                                                }))
                        .collect(Collectors.toList());

        int fetchThreadNum = 100;
        CountDownLatch latch = new CountDownLatch(fetchThreadNum);
        ResultFetcher fetcher = buildResultFetcher(dataSuppliers, 1);
        Map<Long, List<RowData>> rows = new ConcurrentHashMap<>();

        AtomicReference<Boolean> payloadHasData = new AtomicReference<>(true);
        for (int i = 0; i < fetchThreadNum; i++) {
            EXECUTOR_EXTENSION
                    .getExecutor()
                    .submit(
                            () -> {
                                ResultSet resultSet =
                                        fetcher.fetchResults(FetchOrientation.FETCH_NEXT, 1);
                                if (resultSet.getResultType().equals(ResultSet.ResultType.PAYLOAD)
                                        && resultSet.getData().isEmpty()) {
                                    payloadHasData.set(false);
                                }

                                rows.compute(
                                        Thread.currentThread().getId(),
                                        (k, v) -> {
                                            if (v == null) {
                                                return resultSet.getData();
                                            } else {
                                                v.addAll(resultSet.getData());
                                                return v;
                                            }
                                        });

                                latch.countDown();
                            });
        }

        latch.await();
        assertEquals(true, payloadHasData.get());
        assertEquals(
                new HashSet<>(data),
                rows.values().stream().flatMap(List::stream).collect(Collectors.toSet()));
    }