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 testFetchFailedResult() {
        String message = "Artificial Exception";
        ResultFetcher fetcher =
                buildResultFetcher(
                        Arrays.asList(TestIterator.createErrorIterator(message), data.iterator()),
                        data.size());

        assertThatThrownBy(
                        () -> {
                            Long token = 0L;
                            while (token != null) {
                                // Use loop to fetch results from the ErrorIterator
                                token =
                                        fetcher.fetchResults(token, Integer.MAX_VALUE)
                                                .getNextToken();
                            }
                        })
                .satisfies(FlinkAssertions.anyCauseMatches(message));
    }