code: def set_consumer_offsets(
      kafka_client,
      group,
      new_offsets,
      raise_on_error=True,
  ):
      
      valid_new_offsets = _verify_commit_offsets_requests(
          kafka_client,
          new_offsets,
          raise_on_error
      )
  
      group_offset_reqs = [
          OffsetCommitRequestPayload(
              topic,
              partition,
              offset,
              metadata='',
          )
          for topic, new_partition_offsets in six.iteritems(valid_new_offsets)
          for partition, offset in six.iteritems(new_partition_offsets)
      ]
  
      send_api = kafka_client.send_offset_commit_request_kafka
  
      status = []
      if group_offset_reqs:
          status = send_api(
              group,
              group_offset_reqs,
              raise_on_error,
              callback=_check_commit_response_error
          )
  
      return [_f for _f in status
              if _f and _f.error != 0]
docs: Set consumer offsets to the specified offsets.
  
      This method does not validate the specified offsets, it is up to
      the caller to specify valid offsets within a topic partition.
  
      If any partition leader is not available, the request fails for all the
      other topics. This is the tradeoff of sending all topic requests in batch
      and save both in performance and Kafka load.
  
      :param kafka_client: a connected KafkaToolClient
      :param group: kafka group_id
      :param topics: dict {<topic>: {<partition>: <offset>}}
      :param raise_on_error: if False the method does not raise exceptions
        on errors encountered. It may still fail on the request send.
      :returns: a list of errors for each partition offset update that failed.
      :rtype: list [OffsetCommitError]
      :raises:
        :py:class:`kafka_utils.util.error.UnknownTopic`: upon missing
        topics and raise_on_error=True
  
        :py:class:`kafka_utils.util.error.UnknownPartition`: upon missing
        partitions and raise_on_error=True
  
        :py:class:`exceptions.TypeError`: upon badly formatted input
        new_offsets
  
        FailedPayloadsError: upon send request error.