'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
InvalidMessageError: 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.'

def fetched_records(self, max_records=None):

    if (max_records is None):
        max_records = self.config['max_poll_records']
    assert (max_records > 0)
    if self._subscriptions.needs_partition_assignment:
        return ({}, False)
    self._raise_if_offset_out_of_range()
    self._raise_if_unauthorized_topics()
    self._raise_if_record_too_large()
    drained = collections.defaultdict(list)
    partial = bool((self._records and max_records))
    while (self._records and (max_records > 0)):
        part = self._records.popleft()
        max_records -= self._append(drained, part, max_records)
        if part.has_more():
            self._records.appendleft(part)
        else:
            partial &= False
    return (dict(drained), partial)
