def fetched_records(self, max_records=None, update_offsets=True):
        """Returns previously fetched records and updates consumed offsets.

        Arguments:
            max_records (int): Maximum number of records returned. Defaults
                to max_poll_records configuration.

        Raises:
            OffsetOutOfRangeError: if no subscription offset_reset_strategy
            CorruptRecordError: if message crc validation fails (check_crcs
                must be set to True)
            RecordTooLargeError: if a message is larger than the currently
                configured max_partition_fetch_bytes
            TopicAuthorizationError: if consumer is not authorized to fetch
                messages from the topic

        Returns: (records (dict), partial (bool))
            records: {TopicPartition: [messages]}
            partial: True if records returned did not fully drain any pending
                partition requests. This may be useful for choosing when to
                pipeline additional fetch requests.
        """
        if max_records is None:
            max_records = self.config['max_poll_records']
        assert max_records > 0

        if self._next_in_line_exception_metadata is not None:
            exc_meta = self._next_in_line_exception_metadata
            self._next_in_line_exception_metadata = None
            tp = exc_meta.partition
            if self._subscriptions.is_fetchable(tp) and self._subscriptions.position(tp).offset == exc_meta.fetched_offset:
                raise exc_meta.exception

        drained = collections.defaultdict(list)
        records_remaining = max_records
        # Needed to construct ExceptionMetadata if any exception is found when processing completed_fetch
        fetched_partition = None
        fetched_offset = -1

        try:
            while records_remaining > 0:
                if not self._next_partition_records:
                    if not self._completed_fetches:
                        break
                    completion = self._completed_fetches.popleft()
                    fetched_partition = completion.topic_partition
                    fetched_offset = completion.fetched_offset
                    self._next_partition_records = self._parse_fetched_data(completion)
                else:
                    fetched_partition = self._next_partition_records.topic_partition
                    fetched_offset = self._next_partition_records.next_fetch_offset
                    records_remaining -= self._append(drained,
                                                      self._next_partition_records,
                                                      records_remaining,
                                                      update_offsets)
        except Exception as e:
            if not drained:
                raise e
            # To be thrown in the next call of this method
            self._next_in_line_exception_metadata = ExceptionMetadata(fetched_partition, fetched_offset, e)
        return dict(drained), bool(self._completed_fetches)
Returns previously fetched records and updates consumed offsets.

        Arguments:
            max_records (int): Maximum number of records returned. Defaults
                to max_poll_records configuration.

        Raises:
            OffsetOutOfRangeError: if no subscription offset_reset_strategy
            CorruptRecordError: if message crc validation fails (check_crcs
                must be set to True)
            RecordTooLargeError: if a message is larger than the currently
                configured max_partition_fetch_bytes
            TopicAuthorizationError: if consumer is not authorized to fetch
                messages from the topic

        Returns: (records (dict), partial (bool))
            records: {TopicPartition: [messages]}
            partial: True if records returned did not fully drain any pending
                partition requests. This may be useful for choosing when to
                pipeline additional fetch requests.
        
kafka/consumer/fetcher.py