@ParameterizedTest
    @MethodSource("partitionDataErrorSupplier")
    public void testOffsetFetchRequestPartitionDataError(final Errors error, final boolean isRetriable) {
        CommitRequestManager commitRequestManager = create(true, 100);
        when(coordinatorRequestManager.coordinator()).thenReturn(Optional.of(mockedNode));
        Set<TopicPartition> partitions = new HashSet<>();
        TopicPartition tp1 = new TopicPartition("t1", 2);
        TopicPartition tp2 = new TopicPartition("t2", 3);
        partitions.add(tp1);
        partitions.add(tp2);
        long deadlineMs = time.milliseconds() + defaultApiTimeoutMs;
        CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> future =
                commitRequestManager.fetchOffsets(partitions, deadlineMs);

        NetworkClientDelegate.PollResult res = commitRequestManager.poll(time.milliseconds());
        assertEquals(1, res.unsentRequests.size());

        // Setting 1 partition with error
        HashMap<TopicPartition, OffsetFetchResponse.PartitionData> topicPartitionData = new HashMap<>();
        topicPartitionData.put(tp1, new OffsetFetchResponse.PartitionData(100L, Optional.of(1), "metadata", error));
        topicPartitionData.put(tp2, new OffsetFetchResponse.PartitionData(100L, Optional.of(1), "metadata", Errors.NONE));

        res.unsentRequests.get(0).handler().onComplete(buildOffsetFetchClientResponse(
                res.unsentRequests.get(0),
                topicPartitionData,
                Errors.NONE,
                false));
        if (isRetriable)
            testRetriable(commitRequestManager, Collections.singletonList(future));
        else
            testNonRetriable(Collections.singletonList(future));
    }
@return Retriable coordinator errors and a boolean indicating if it is expected to mark
the coordinator unknown when the error occurs.
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java