@ParameterizedTest
    @MethodSource("handleFetchResponseErrorSupplier")
    public void testHandleFetchResponseError(Errors error,
                                             long highWatermark,
                                             boolean hasTopLevelError,
                                             boolean shouldRequestMetadataUpdate) {
        buildFetcher();
        assignFromUser(singleton(tp0));
        subscriptions.seek(tp0, 0);

        assertEquals(1, sendFetches());

        final FetchResponse fetchResponse;

        if (hasTopLevelError)
            fetchResponse = fetchResponseWithTopLevelError(tidp0, error, 0);
        else
            fetchResponse = fullFetchResponse(tidp0, records, error, highWatermark, 0);

        client.prepareResponse(fetchResponse);
        consumerClient.poll(time.timer(0));

        assertEmptyFetch("Should not return records or advance position on fetch error");

        long timeToNextUpdate = metadata.timeToNextUpdate(time.milliseconds());

        if (shouldRequestMetadataUpdate)
            assertEquals(0L, timeToNextUpdate, "Should have requested metadata update");
        else
            assertNotEquals(0L, timeToNextUpdate, "Should not have requested metadata update");
    }
Test the case where the client makes a post KIP-74 FetchRequest, but the server replies with only a
partial request. For v3 and later FetchRequests, the implementation of KIP-74 changed the behavior
so that at least one message is always returned. Therefore, this case should not happen, and it indicates
that an internal error has taken place.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/FetcherTest.java