with self._thread_lock:
            # Inject the existing future so _on_response resolves it.
            self._queue.put((initial_user_request, initial_future))

            merged = self._make_initial_request(initial_user_request)
            metadata = tuple(self._metadata) + (
                (
                    "x-goog-request-params",
                    f"write_stream={self._stream_name}",
                ),
            )
            rpc = bidi.BidiRpc(
                self._client.append_rows,
                initial_request=merged,
                metadata=metadata,
            )
            rpc.add_done_callback(self._on_rpc_done)

            consumer = bidi.BackgroundConsumer(rpc, self._on_response)
            consumer.start()

            self._rpc = rpc
            self._consumer = consumer

        start_time = time.monotonic()
        while not rpc.is_active and consumer.is_active:
            time.sleep(_WRITE_OPEN_INTERVAL)
            if timeout is not None and time.monotonic() - start_time > timeout:
                break

        if not consumer.is_active:
            # Connection failed — drain the queue and fail futures directly
            # rather than going through close() to avoid triggering another
            # reconnect attempt (which would cause an infinite retry loop).
            exc = exceptions.Unknown(
                "There was a problem reopening the stream after a transient error. "
                "Try turning on DEBUG level logs to see the error."
            )
            with self._thread_lock:
                self._closed = True
                while not self._queue.empty():
                    _, future = self._queue.get_nowait()
                    if not future.done():
                        future.set_exception(exc)
                for _, future in pending:
                    if not future.done():
                        future.set_exception(exc)
            return