docs: def _poll_once(self, timeout_ms, max_records):
          """Do one round of polling. In addition to checking for new data, this does
          any needed heart-beating, auto-commits, and offset updates.
  
          Arguments:
              timeout_ms (int): The maximum time in milliseconds to block.
  
          Returns:
              dict: Map of topic to list of records (may be empty).
          """
          self._coordinator.poll()
  
          # Fetch positions if we have partitions we're subscribed to that we
          # don't know the offset for
          if not self._subscription.has_all_fetch_positions():
              self._update_fetch_positions(self._subscription.missing_fetch_positions())
  
          # If data is available already, e.g. from a previous network client
          # poll() call to commit, then just return it immediately
          records, partial = self._fetcher.fetched_records(max_records)
          if records:
              # Before returning the fetched records, we can send off the
              # next round of fetches and avoid block waiting for their
              # responses to enable pipelining while the user is handling the
              # fetched records.
              if not partial:
                  self._fetcher.send_fetches()
              return records
  
          # Send any new fetches (won't resend pending fetches)
          self._fetcher.send_fetches()
  
          timeout_ms = min(timeout_ms, self._coordinator.time_to_next_poll() * 1000)
          self._client.poll(timeout_ms=timeout_ms)
          # after the long poll, we should check whether the group needs to rebalance
          # prior to returning data so that the group can stabilize faster
          if self._coordinator.need_rejoin():
              return {}
  
          records, _ = self._fetcher.fetched_records(max_records)
          return records