docs: def _get_kafka_consumer_offsets(self, instance, consumer_groups):
          """
          retrieve consumer offsets via the new consumer api. Offsets in this version are stored directly
          in kafka (__consumer_offsets topic) rather than in zookeeper
          """
          consumer_offsets = {}
          topics = defaultdict(set)
  
          cli = self._get_kafka_client(instance)
  
          for consumer_group, topic_partitions in iteritems(consumer_groups):
              try:
                  coordinator_id = self._get_group_coordinator(cli, consumer_group)
                  if coordinator_id:
                      offsets = self._get_consumer_offsets(cli, consumer_group, topic_partitions, coordinator_id)
                  else:
                      offsets = self._get_consumer_offsets(cli, consumer_group, topic_partitions)
                      self.log.info("unable to find group coordinator for %s", consumer_group)
  
                  for (topic, partition), offset in iteritems(offsets):
                      topics[topic].update([partition])
                      key = (consumer_group, topic, partition)
                      consumer_offsets[key] = offset
              except Exception:
                  self.log.exception('Could not read consumer offsets from kafka.')
  
          return consumer_offsets, topics