def fetch(self, max_size=None, retry_limit=3, ignore_failures=False):
        """Return a FetchResult, which can be iterated over as a list of 
        MessageSets. A MessageSet is returned for every broker partition that
        is successfully queried, even if that MessageSet is empty.

        FIXME: This is where the adjustment needs to happen. Regardless of 
        whether a rebalance has occurred or not, we can very easily see if we
        are still responsible for the same partitions as we were the last time
        we ran, and set self._bps_to_next_offsets --> we just need to check if
        it's not None and if we still have the same offsets, and adjust 
        accordingly.
        """
        def needs_offset_values_from_zk(bps_to_offsets):
            """We need to pull offset values from ZK if we have no 
            BrokerPartitions in our BPs -> Offsets mapping, or if some of those
            Offsets are unknown (None)"""
            return (not bps_to_offsets) or (None in bps_to_offsets.values())

        log.debug("Fetch called on ZKConsumer {0}".format(self.id))
        if self._needs_rebalance:
            self.rebalance()

        # Find where we're starting from. If we've already done a fetch, we use 
        # our internal value. This is also all we can do in the case where 
        # autocommit is off, since any value in ZK will be out of date.
        bps_to_offsets = dict(self._bps_to_next_offsets)
        offsets_pulled_from_zk = False

        if needs_offset_values_from_zk(bps_to_offsets):
            # We have some offsets, but we've been made responsible for new
            # BrokerPartitions that we need to lookup.
            if bps_to_offsets:
                bps_needing_offsets = [bp for bp, offset in bps_to_offsets.items() 
                                       if offset is None]
            # Otherwise, it's our first fetch, so we need everything
            else:
                bps_needing_offsets = self.broker_partitions

            bps_to_offsets.update(self._zk_util.offsets_for(self.consumer_group,
                                                            self._id,
                                                            bps_needing_offsets))
            offsets_pulled_from_zk = True

        # Do all the fetches we need to (this should get replaced with 
        # multifetch or performance is going to suck wind later)...
        message_sets = []
        # We only iterate over those broker partitions for which we have offsets
        for bp in bps_to_offsets:
            offset = bps_to_offsets[bp]
            kafka = self._connections[bp.broker_id]
            partition = kafka.partition(bp.topic, bp.partition)

            if offset is None:
                offset = partition.latest_offset()
            
            try:
                offsets_msgs = kafka.fetch(bp.topic, 
                                           offset,
                                           partition=bp.partition,
                                           max_size=max_size)

            # If our fetch fails because it's out of range, and the values came
            # from ZK originally (not our internal incrementing), we assume ZK
            # is somehow stale, so we just grab the latest and march on.
            except OffsetOutOfRange as ex:
                if offsets_pulled_from_zk:
                    log.error("Offset {0} from ZooKeeper is out of range for {1}"
                              .format(offset, bp))
                    offset = partition.latest_offset()
                    log.error("Retrying with offset {0} for {1}"
                              .format(offset, bp))
                    offsets_msgs = kafka.fetch(bp.topic, 
                                               offset,
                                               partition=bp.partition,
                                               max_size=max_size)
                else:
                    raise
            except KafkaError as k_err:
                if ignore_failures:
                    log.error("Ignoring failed fetch on {0}".format(bp))
                    log.exception(k_err)
                    continue
                else:
                    raise

            msg_set = MessageSet(bp, offset, offsets_msgs)

            # fetches bytes messages max_fetch
            old_stats = self._stats[bp]
            self._stats[bp] = ConsumerStats(fetches=old_stats.fetches + 1,
                                        bytes=old_stats.bytes + msg_set.size,
                                        messages=old_stats.messages + len(msg_set),
                                        max_fetch=max(old_stats.max_fetch, msg_set.size))

            message_sets.append(msg_set)
        
        result = FetchResult(sorted(message_sets))

        # Now persist our new offsets
        for msg_set in result:
            self._bps_to_next_offsets[msg_set.broker_partition] = msg_set.next_offset

        if self._autocommit:
            self.commit_offsets()

        return result
dataset/ETHPy150Open DataDog/brod/brod/zk.py/ZKConsumer.fetch/original