Unclosed AIOKafkaConnection that does not cause the consumer to crash about aiokafka

[YraganTron]

It seems that the only place where the connection is deleted until the consumer is running is the exit from the method

async def start(self):
    self._client.bootstrap()
    # Here

The next assumption is that when the connection's closing method is called, the connection is not closed immediately. Related comment

    def close(self, reason=None, exc=None):
        self.log.debug("Closing connection at %s:%s", self._host, self._port)
        if self._reader is not None:
            self._writer.close()
            self._writer = self._reader = None
            if not self._read_task.done():
                self._read_task.cancel()
                self._read_task = None
            for _, _, fut in self._requests:
                if not fut.done():
                    error = Errors.KafkaConnectionError(
                        f"Connection at {self._host}:{self._port} closed"
                    )
                    if exc is not None:
                        error.__cause__ = exc
                        error.__context__ = exc
                    fut.set_exception(error)
            self._requests = collections.deque()
            if self._on_close_cb is not None:
                self._on_close_cb(self, reason)
                self._on_close_cb = None
        if self._idle_handle is not None:
            self._idle_handle.cancel()

        # transport.close() will close socket, but not right ahead. Return
        # a future in case we need to wait on it.
        return self._closed_fut

We have a workaround with a future self._closed_fut, waiting for which we guarantee that the socket was closed. It seems we are missing waiting for this future when calling bootstrap method