docs: def _send_offset_commit_request(self, offsets):
          """Commit offsets for the specified list of topics and partitions.
  
          This is a non-blocking call which returns a request future that can be
          polled in the case of a synchronous commit or ignored in the
          asynchronous case.
  
          Arguments:
              offsets (dict of {TopicPartition: OffsetAndMetadata}): what should
                  be committed
  
          Returns:
              Future: indicating whether the commit was successful or not
          """
          assert self.config['api_version'] >= (0, 8, 1), 'Unsupported Broker API'
          assert all(map(lambda k: isinstance(k, TopicPartition), offsets))
          assert all(map(lambda v: isinstance(v, OffsetAndMetadata),
                         offsets.values()))
          if not offsets:
              log.debug('No offsets to commit')
              return Future().success(None)
  
          node_id = self.coordinator()
          if node_id is None:
              return Future().failure(Errors.GroupCoordinatorNotAvailableError)
  
  
          # create the offset commit request
          offset_data = collections.defaultdict(dict)
          for tp, offset in six.iteritems(offsets):
              offset_data[tp.topic][tp.partition] = offset
  
          if self._subscription.partitions_auto_assigned():
              generation = self.generation()
          else:
              generation = Generation.NO_GENERATION
  
          # if the generation is None, we are not part of an active group
          # (and we expect to be). The only thing we can do is fail the commit
          # and let the user rejoin the group in poll()
          if self.config['api_version'] >= (0, 9) and generation is None:
              return Future().failure(Errors.CommitFailedError())
  
          if self.config['api_version'] >= (0, 9):
              request = OffsetCommitRequest[2](
                  self.group_id,
                  generation.generation_id,
                  generation.member_id,
                  OffsetCommitRequest[2].DEFAULT_RETENTION_TIME,
                  [(
                      topic, [(
                          partition,
                          offset.offset,
                          offset.metadata
                      ) for partition, offset in six.iteritems(partitions)]
                  ) for topic, partitions in six.iteritems(offset_data)]
              )
          elif self.config['api_version'] >= (0, 8, 2):
              request = OffsetCommitRequest[1](
                  self.group_id, -1, '',
                  [(
                      topic, [(
                          partition,
                          offset.offset,
                          -1,
                          offset.metadata
                      ) for partition, offset in six.iteritems(partitions)]
                  ) for topic, partitions in six.iteritems(offset_data)]
              )
          elif self.config['api_version'] >= (0, 8, 1):
              request = OffsetCommitRequest[0](
                  self.group_id,
                  [(
                      topic, [(
                          partition,
                          offset.offset,
                          offset.metadata
                      ) for partition, offset in six.iteritems(partitions)]
                  ) for topic, partitions in six.iteritems(offset_data)]
              )
  
          log.debug("Sending offset-commit request with %s for group %s to %s",
                    offsets, self.group_id, node_id)
  
          future = Future()
          _f = self._client.send(node_id, request)
          _f.add_callback(self._handle_offset_commit_response, offsets, future, time.time())
          _f.add_errback(self._failed_request, node_id, request, future)
          return future