Fetcher { public void sendFetches() { Map<Node, FetchRequest> fetchRequests = createFetchRequests(); for (Map.Entry<Node, FetchRequest> fetchEntry : fetchRequests.entrySet()) { final FetchRequest request = fetchEntry.getValue(); client.send(fetchEntry.getKey(), ApiKeys.FETCH, request) .addListener(new RequestFutureListener<ClientResponse>() { @Override public void onSuccess(ClientResponse resp) { FetchResponse response = new FetchResponse(resp.responseBody()); Set<TopicPartition> partitions = new HashSet<>(response.responseData() .keySet()); FetchResponseMetricAggregator metricAggregator = new FetchResponseMetricAggregator(sensors, partitions); for (Map.Entry<TopicPartition, FetchResponse.PartitionData> entry : response.responseData() .entrySet()) { TopicPartition partition = entry.getKey(); FetchResponse.PartitionData fetchData = entry.getValue(); long fetchOffset = request.fetchData() .get(partition).offset; completedFetches.add(new CompletedFetch(partition, fetchOffset, fetchData, metricAggregator)); } sensors.fetchLatency.record(resp.requestLatencyMs()); sensors.fetchThrottleTimeSensor.record(response.getThrottleTime()); } @Override public void onFailure(RuntimeException e) { log.debug("Fetch failed", e); } }); } } Fetcher(ConsumerNetworkClient client,
        int minBytes,
        int maxWaitMs,
        int fetchSize,
        int maxPollRecords,
        boolean checkCrcs,
        Deserializer<K> keyDeserializer,
        Deserializer<V> valueDeserializer,
        Metadata metadata,
        SubscriptionState subscriptions,
        Metrics metrics,
        String metricGrpPrefix,
        Time time,
        long retryBackoffMs); void sendFetches(); Map<TopicPartition, List<ConsumerRecord<K, V>>> fetchedRecords(); void updateFetchPositions(Set<TopicPartition> partitions); Map<String, List<PartitionInfo>> getAllTopicMetadata(long timeout); Map<String, List<PartitionInfo>> getTopicMetadata(MetadataRequest request, long timeout);  }
@Test public void testFetchOnPausedPartition() { subscriptions.assignFromUser(Arrays.asList(tp)); subscriptions.seek(tp, 0); subscriptions.pause(tp); fetcher.sendFetches(); assertTrue(client.requests().isEmpty()); }