'Iterate over fetched_records'

def _message_generator(self):

    if self._subscriptions.needs_partition_assignment:
        raise StopIteration('Subscription    needs    partition    assignment')
    while self._records:
        self._raise_if_offset_out_of_range()
        self._raise_if_unauthorized_topics()
        self._raise_if_record_too_large()
        if (len(self._records) <= self.config['iterator_refetch_records']):
            self.send_fetches()
        part = self._records.popleft()
        tp = part.topic_partition
        fetch_offset = part.fetch_offset
        if (not self._subscriptions.is_assigned(tp)):
            log.debug('Not    returning    fetched    records    for    partition    %s    since    it    is    no    longer    assigned', tp)
            continue
        position = self._subscriptions.assignment[tp].position
        if (not self._subscriptions.is_fetchable(tp)):
            log.debug('Not    returning    fetched    records    for    assigned    partition    %s    since    it    is    no    longer    fetchable', tp)
        elif (fetch_offset == position):
            log.log(0, 'Returning    fetched    records    at    offset    %d    for    assigned    partition    %s', position, tp)
            self._subscriptions.assignment[tp].drop_pending_message_set = False
            for msg in part.messages:
                if (not self._subscriptions.is_fetchable(tp)):
                    log.debug('Not    returning    fetched    records    for    partition    %s    since    it    is    no    longer    fetchable', tp)
                    break
                elif self._subscriptions.assignment[tp].drop_pending_message_set:
                    log.debug('Skipping    remainder    of    message    set    for    partition    %s', tp)
                    self._subscriptions.assignment[tp].drop_pending_message_set = False
                    break
                elif (msg.offset < self._subscriptions.assignment[tp].position):
                    log.debug('Skipping    message    offset:    %s    (expecting    %s)', msg.offset, self._subscriptions.assignment[tp].position)
                    continue
                self._subscriptions.assignment[tp].position = (msg.offset + 1)
                (yield msg)
        else:
            log.debug('Ignoring    fetched    records    for    %s    at    offset    %s    since    the    current    position    is    %d', tp, part.fetch_offset, position)
