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 testFetchIllegalToken() {
        ResultFetcher fetcher =
                buildResultFetcher(Collections.singletonList(data.iterator()), data.size());
        assertThatThrownBy(() -> fetcher.fetchResults(2, Integer.MAX_VALUE))
                .satisfies(FlinkAssertions.anyCauseMatches("Expecting token to be 0, but found 2"));
    }