def poll(self, timeout_ms=None, future=None):
        """Try to read and write to sockets.

        This method will also attempt to complete node connections, refresh
        stale metadata, and run previously-scheduled tasks.

        Arguments:
            timeout_ms (int, optional): maximum amount of time to wait (in ms)
                for at least one response. Must be non-negative. The actual
                timeout will be the minimum of timeout, request timeout and
                metadata timeout. Default: request_timeout_ms
            future (Future, optional): if provided, blocks until future.is_done

        Returns:
            list: responses received (can be empty)
        """
        if not isinstance(timeout_ms, (int, float, type(None))):
            raise TypeError('Invalid type for timeout: %s' % type(timeout_ms))
        timer = Timer(timeout_ms)

        # Loop for futures, break after first loop if None
        responses = []
        while True:
            with self._lock:
                if self._closed:
                    break

                # Attempt to complete pending connections
                for node_id in list(self._connecting):
                    # False return means no more connection progress is possible
                    # Connected nodes will update _connecting via state_change callback
                    if not self._init_connect(node_id):
                        # It's possible that the connection attempt triggered a state change
                        # but if not, make sure to remove from _connecting list
                        if node_id in self._connecting:
                            self._connecting.remove(node_id)

                # Send a metadata request if needed (or initiate new connection)
                metadata_timeout_ms = self._maybe_refresh_metadata()

                # If we got a future that is already done, don't block in _poll
                if future is not None and future.is_done:
                    timeout = 0
                else:
                    user_timeout_ms = timer.timeout_ms if timeout_ms is not None else self.config['request_timeout_ms']
                    idle_connection_timeout_ms = self._idle_expiry_manager.next_check_ms()
                    request_timeout_ms = self._next_ifr_request_timeout_ms()
                    log.debug("Timeouts: user %f, metadata %f, idle connection %f, request %f", user_timeout_ms, metadata_timeout_ms, idle_connection_timeout_ms, request_timeout_ms)
                    timeout = min(
                        user_timeout_ms,
                        metadata_timeout_ms,
                        idle_connection_timeout_ms,
                        request_timeout_ms)
                    timeout = max(0, timeout)  # avoid negative timeouts

                self._poll(timeout / 1000)

            # called without the lock to avoid deadlock potential
            # if handlers need to acquire locks
            responses.extend(self._fire_pending_completed_requests())

            # If all we had was a timeout (future is None) - only do one poll
            # If we do have a future, we keep looping until it is done
            if future is None:
                break
            elif future.is_done:
                break
            elif timeout_ms is not None and timer.expired:
                break

        return responses
Try to read and write to sockets.

        This method will also attempt to complete node connections, refresh
        stale metadata, and run previously-scheduled tasks.

        Arguments:
            timeout_ms (int, optional): maximum amount of time to wait (in ms)
                for at least one response. Must be non-negative. The actual
                timeout will be the minimum of timeout, request timeout and
                metadata timeout. Default: request_timeout_ms
            future (Future, optional): if provided, blocks until future.is_done

        Returns:
            list: responses received (can be empty)
        
kafka/client_async.py