private void sendOffsetFetchResponse(MockClient mockClient, Map<String, ListConsumerGroupOffsetsSpec> groupSpecs, boolean batched, Errors error) throws Exception {
        waitForRequest(mockClient, ApiKeys.OFFSET_FETCH);

        ClientRequest clientRequest = mockClient.requests().peek();
        OffsetFetchRequestData data = ((OffsetFetchRequest.Builder) clientRequest.requestBuilder()).data;
        Map<String, Map<TopicPartition, PartitionData>> results = new HashMap<>();
        Map<String, Errors> errors = new HashMap<>();
        data.groups().forEach(group -> {
            Map<TopicPartition, PartitionData> partitionResults = new HashMap<>();
            for (TopicPartition tp : groupSpecs.get(group.groupId()).topicPartitions()) {
                partitionResults.put(tp, new PartitionData(10, Optional.empty(), "", Errors.NONE));
            }
            results.put(group.groupId(), partitionResults);
            errors.put(group.groupId(), error);
        });
        if (!batched) {
            assertEquals(1, data.groups().size());
            mockClient.respond(new OffsetFetchResponse(THROTTLE, error, results.values().iterator().next()));
        } else
            mockClient.respond(new OffsetFetchResponse(THROTTLE, errors, results));
    }
Test that we propagate exceptions encountered when fetching metadata.
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java