@Test
    public void testFetchWithOffsetOutOfRange() {
        buildDependencies();
        assignAndSeek(topicAPartition0);

        CompletedFetch completedFetch = completedFetchBuilder.build();
        fetchBuffer.add(completedFetch);

        // Fetch the data and validate that we get our first batch of records back.
        Fetch<String, String> fetch = fetchCollector.collectFetch(fetchBuffer);
        assertFalse(fetch.isEmpty());
        assertEquals(DEFAULT_RECORD_COUNT, fetch.numRecords());

        // Try to fetch more data and validate that we get an empty Fetch back.
        completedFetch = completedFetchBuilder
                .fetchOffset(fetch.numRecords())
                .error(Errors.OFFSET_OUT_OF_RANGE)
                .build();
        fetchBuffer.add(completedFetch);
        fetch = fetchCollector.collectFetch(fetchBuffer);
        assertTrue(fetch.isEmpty());

        // Try to fetch more data and validate that we get an empty Fetch back.
        completedFetch = completedFetchBuilder
                .fetchOffset(fetch.numRecords())
                .error(Errors.OFFSET_OUT_OF_RANGE)
                .build();
        fetchBuffer.add(completedFetch);
        fetch = fetchCollector.collectFetch(fetchBuffer);
        assertTrue(fetch.isEmpty());
    }
This tests the {@link FetchCollector} functionality in addition to what {@link FetcherTest} tests during the course
of its tests.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetchCollectorTest.java