Making topic write stream creation non-blocking

[Myllyenko]

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.

Compatibility

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

Testing

Three new tests in TopicRetryableStreamTest cover the states the change introduces:

• asyncStreamCreationTest — start() returns before the underlying gRPC stream is started, and messages sent in the meantime are skipped
• closeWhileStreamIsCreatingTest — a stream created after close() is never started
• streamCreationFailedTest — a failed creation surfaces as CLIENT_INTERNAL_ERROR through the normal stop path

[codecov]

Codecov Report

❌ Patch coverage is 87.17949% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.08%. Comparing base (aa4ad05) to head (ecb00ef).
⚠️ Report is 5 commits behind head on master.

Files with missing lines	Patch %	Lines
...ydb/topic/write/impl/WriteStreamDirectFactory.java	84.00%	3 Missing and 1 partial ⚠️
...java/tech/ydb/topic/impl/TopicRetryableStream.java	92.30%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #720      +/-   ##
============================================
+ Coverage     73.02%   73.08%   +0.06%     
- Complexity     3562     3575      +13     
============================================
  Files           391      391              
  Lines         16497    16517      +20     
  Branches       1730     1725       -5     
============================================
+ Hits          12047    12072      +25     
+ Misses         3828     3824       -4     
+ Partials        622      621       -1     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.