Problem

TopicRetryableStream.start() runs on the transport's shared scheduler on every reconnect. With directWrite enabled, WriteStreamDirectFactory.createNewStream resolved the target partition and its location synchronously inside that call:

• lookupPartitionId() — join() on a probe-stream future (1 min deadline)
• lookupLocation() — join() on describeTopic() (1 min deadline)

So a single reconnect toward an unresponsive destination could hold a scheduler thread for up to two minutes.

That scheduler is sized max(cores / 2, 2) and is shared with discovery, the table/query session pools, retry contexts and OperationTray. A handful of stalled direct writers is enough to exhaust it head-of-line: session-acquire timeouts stop firing, discovery ticks stop running, and the whole transport degrades — not just the topic writers that caused it.

Change

createNewStream now returns CompletableFuture and the two lookups are composed rather than joined, so no shared scheduler thread is ever held while a stream is being created.

Behavior notes for reviewers

• The continuation moves off the scheduler. describeTopic, rpc.writeSession(...) and the init sendNext now run on whichever thread completes the discovery future (a gRPC callback thread) instead of a scheduler thread. That is the point of the change; ReadWriteStreamCall guards sendNext/close/cancel with a ReentrantLock, so the relocation is safe.
• Probe-stream cleanup is unchanged in effect. The old finally { if (!streamFuture.isDone()) stream.close(); } became a whenComplete on the partition-id future, registered before the init is sent, so it still fires on both the success path and the sendNext-throws path.
• An un-started stream is not closed on the "closed while creating" path. This matches the existing convention in the class — doubleStartTest already asserts the surplus stream is neither started nor closed — and ReadWriteStreamCall's constructor starts no RPC, so nothing leaks.
• onClose/onRetry can now receive a null stream. When creation itself fails there is no stream to report, so onStreamStop(null, ...) is used. onStreamStop only passes the argument through to onClose/onRetry and never dereferences it, and WriteSession — the only subclass — ignores the parameter in both. The two abstract methods are documented accordingly.
• Closing during creation does not deliver onClose. On that path close() finds no registered stream and returns false. WriterImpl.shutdown() already handles exactly that case ("stream will never call onClose") by completing shutdownFuture itself, so no shutdown hangs.

Compatibility

Source-incompatible for anyone subclassing TopicRetryableStream or WriteStreamFactory: createNewStream changed its return type, from S to CompletableFuture<S>. Both are internal impl classes, and inside the repo the only subclass is WriteSession. Should go into a minor release, not a patch.