'Fetch data from assigned topics / partitions.
Records are fetched and returned in batches by topic-partition.
On each poll, consumer will try to use the last consumed offset as the
starting offset and fetch sequentially. The last consumed offset can be
manually set through :meth:`~kafka.KafkaConsumer.seek` or automatically
set as the last committed offset for the subscribed list of partitions.
Incompatible with iterator interface -- use one or the other, not both.
Arguments:
timeout_ms (int, optional): Milliseconds spent waiting in poll if
data is not available in the buffer. If 0, returns immediately
with any records that are available currently in the buffer,
else returns empty. Must not be negative. Default: 0
max_records (int, optional): The maximum number of records returned
in a single call to :meth:`~kafka.KafkaConsumer.poll`.
Default: Inherit value from max_poll_records.
Returns:
dict: Topic to list of records since the last fetch for the
subscribed list of topics and partitions.'

def poll(self, timeout_ms=0, max_records=None):

    assert (timeout_ms >= 0), 'Timeout    must    not    be    negative'
    if (max_records is None):
        max_records = self.config['max_poll_records']
    start = time.time()
    remaining = timeout_ms
    while True:
        records = self._poll_once(remaining, max_records)
        if records:
            return records
        elapsed_ms = ((time.time() - start) * 1000)
        remaining = (timeout_ms - elapsed_ms)
        if (remaining <= 0):
            return {}
