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
retrieve consumer offsets via the new consumer api. Offsets in this version are stored directly
        in kafka (__consumer_offsets topic) rather than in zookeeper