code: 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']
          assert isinstance(max_records, int), 'max_records must be an integer'
          assert max_records > 0, 'max_records must be positive'
  
          # Poll for new data until the timeout expires
          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 {}
docs: 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.