'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).'

def _poll_once(self, timeout_ms, max_records):

    if self._use_consumer_group():
        self._coordinator.ensure_coordinator_known()
        self._coordinator.ensure_active_group()
    elif ((self.config['group_id'] is not None) and (self.config['api_version'] >= (0, 8, 2))):
        self._coordinator.ensure_coordinator_known()
    if (not self._subscription.has_all_fetch_positions()):
        self._update_fetch_positions(self._subscription.missing_fetch_positions())
    (records, partial) = self._fetcher.fetched_records(max_records)
    if records:
        if (not partial):
            self._fetcher.send_fetches()
        return records
    self._fetcher.send_fetches()
    self._client.poll(timeout_ms=timeout_ms, sleep=True)
    (records, _) = self._fetcher.fetched_records(max_records)
    return records
