code: def fetch_committed_offsets(self, partitions):
          
          if not partitions:
              return {}
  
          while True:
              self.ensure_coordinator_ready()
  
              # contact coordinator to fetch committed offsets
              future = self._send_offset_fetch_request(partitions)
              self._client.poll(future=future)
  
              if future.succeeded():
                  return future.value
  
              if not future.retriable():
                  raise future.exception # pylint: disable-msg=raising-bad-type
  
              time.sleep(self.config['retry_backoff_ms'] / 1000)
docs: Fetch the current committed offsets for specified partitions
  
          Arguments:
              partitions (list of TopicPartition): partitions to fetch
  
          Returns:
              dict: {TopicPartition: OffsetAndMetadata}