class_docstr: This class manages the coordination process with the consumer coordinator.
docstr: Fetch the committed offsets for a set of partitions.
  
  This is a non-blocking call. The returned future can be polled to get
  the actual offsets returned from the broker.
  
  Arguments:
      partitions (list of TopicPartition): the partitions to fetch
  
  Returns:
      Future: resolves to dict of offsets: {TopicPartition: OffsetAndMetadata}
docstr_tok: ["Fetch","the","committed","offsets","for","a","set","of","partitions",".","This","is","a","non-blocking","call",".","The","returned","future","can","be","polled","to","get","the","actual","offsets","returned","from","the","broker",".","Arguments",":","partitions","(","list","of","TopicPartition",")",":","the","partitions","to","fetch","Returns",":","Future",":","resolves","to","dict","of","offsets",":","{","TopicPartition",":","OffsetAndMetadata","}"]
code: def _send_offset_fetch_request(self, partitions):
      """Fetch the committed offsets for a set of partitions.
  
      This is a non-blocking call. The returned future can be polled to get
      the actual offsets returned from the broker.
  
      Arguments:
          partitions (list of TopicPartition): the partitions to fetch
  
      Returns:
          Future: resolves to dict of offsets: {TopicPartition: OffsetAndMetadata}
      """
      assert self.config["api_version"] >= (
          0,
          8,
          1,
      ), "Unsupported Broker API"
      assert all(
          map(lambda k: isinstance(k, TopicPartition), partitions)
      )
      if not partitions:
          return Future().success({})
  
      node_id = self.coordinator()
      if node_id is None:
          return Future().failure(
              Errors.GroupCoordinatorNotAvailableError
          )
  
      # Verify node is ready
      if not self._client.ready(node_id):
          log.debug(
              "Node %s not ready -- failing offset fetch request",
              node_id,
          )
          return Future().failure(Errors.NodeNotReadyError)
  
      log.debug(
          "Group %s fetching committed offsets for partitions: %s",
          self.group_id,
          partitions,
      )
      # construct the request
      topic_partitions = collections.defaultdict(set)
      for tp in partitions:
          topic_partitions[tp.topic].add(tp.partition)
  
      if self.config["api_version"] >= (0, 8, 2):
          request = OffsetFetchRequest[1](
              self.group_id, list(topic_partitions.items())
          )
      else:
          request = OffsetFetchRequest[0](
              self.group_id, list(topic_partitions.items())
          )
  
      # send the request with a callback
      future = Future()
      _f = self._client.send(node_id, request)
      _f.add_callback(self._handle_offset_fetch_response, future)
      _f.add_errback(self._failed_request, node_id, request, future)
      return future
  
imports: from __future__ import absolute_import, division
  import collections
  import copy
  import functools
  import logging
  import time
  from kafka.vendor import six
  from kafka.coordinator.base import BaseCoordinator, Generation
  from kafka.coordinator.assignors.range import RangePartitionAssignor
  from kafka.coordinator.assignors.roundrobin import RoundRobinPartitionAssignor
  from kafka.coordinator.assignors.sticky.sticky_assignor import StickyPartitionAssignor
  from kafka.coordinator.protocol import ConsumerProtocol
  import kafka.errors
  from kafka.future import Future
  from kafka.metrics import AnonMeasurable
  from kafka.metrics.stats import Avg, Count, Max, Rate
  from kafka.protocol.commit import OffsetCommitRequest, OffsetFetchRequest
  from kafka.structs import OffsetAndMetadata, TopicPartition
  from kafka.util import WeakMethod
label_desc: Use image node_id 25 for calling the ConsumerCoordinator obj's underlying member method code with example usage: obj._send_offset_fetch_request(partitions) and returns: future, Future, Future, Future