code: def get_current_consumer_offsets(
      kafka_client,
      group,
      topics,
      raise_on_error=True,
  ):
      
  
      topics = _verify_topics_and_partitions(kafka_client, topics, raise_on_error)
  
      group_offset_reqs = [
          OffsetFetchRequestPayload(topic, partition)
          for topic, partitions in six.iteritems(topics)
          for partition in partitions
      ]
  
      group_offsets = {}
  
      send_api = kafka_client.send_offset_fetch_request_kafka
  
      if group_offset_reqs:
          # fail_on_error = False does not prevent network errors
          group_resps = send_api(
              group=group,
              payloads=group_offset_reqs,
              fail_on_error=False,
              callback=pluck_topic_offset_or_zero_on_unknown,
          )
          for resp in group_resps:
              group_offsets.setdefault(
                  resp.topic,
                  {},
              )[resp.partition] = resp.offset
  
      return group_offsets
docs: Get current consumer offsets.
  
      NOTE: This method does not refresh client metadata. It is up to the caller
      to avoid using stale metadata.
  
      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: topic list or dict {<topic>: [partitions]}
      :param raise_on_error: if False the method ignores missing topics and
        missing partitions. It still may fail on the request send.
      :returns: a dict topic: partition: offset
      :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
  
        FailedPayloadsError: upon send request error.