class Fetcher {

    public void sendFetches() {
        Map<Node, FetchRequest> fetchRequests = createFetchRequests();

        for (Map.Entry<Node, FetchRequest/* 在createFetchRequests里生成的，最主要的是包括了position信息 */> 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();

                              // 这个是原来的offset（提交时的那个）
                              // 也就是 long position = this.subscriptions.position(partition);
                              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);
                      }
                  });
        }
    }

    public  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);

}