Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #720 +/- ##
============================================
- Coverage 73.18% 73.13% -0.06%
- Complexity 3589 3597 +8
============================================
Files 392 392
Lines 16531 16551 +20
Branches 1736 1731 -5
============================================
+ Hits 12099 12104 +5
- Misses 3811 3820 +9
- Partials 621 627 +6 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
TopicRetryableStream.start() is called from the shared transport scheduler on every reconnect. With directWrite enabled, WriteStreamDirectFactory resolved the target partition and its location synchronously inside createNewStream: lookupPartitionId() joined a probe stream future (1 min deadline) and lookupLocation() joined describeTopic() (1 min deadline). Each reconnect of an unresponsive destination could therefore occupy a scheduler thread for up to two minutes. The shared scheduler is sized max(cores / 2, 2) and is also used by discovery, session pools, retry contexts and operation tray, so a handful of stalled writers could stall the whole transport: session acquire timeouts stop firing and discovery ticks stop running. Make createNewStream() return CompletableFuture and compose the partition and location lookups instead of joining them, so no shared scheduler thread is held while a stream is being created. Since stream creation is now asynchronous, close() may happen while it is in progress. TopicRetryableStream handles that by re-checking isClosed after publishing the new stream: close() sets the volatile flag before clearing the stream reference, so a creation that wins the race always observes the flag and drops the stream without starting it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Myllyenko
force-pushed
the
topic-async-write-stream-factory
branch
from
September 17, 2026 15:33
ecb00ef to
8840c95
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
TopicRetryableStream.start()runs on the transport's shared scheduler on every reconnect. WithdirectWriteenabled,WriteStreamDirectFactory.createNewStreamresolved the target partition and its location synchronously inside that call:lookupPartitionId()—join()on a probe-stream future (1 min deadline)lookupLocation()—join()ondescribeTopic()(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 andOperationTray. 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
createNewStreamnow returnsCompletableFutureand 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
describeTopic,rpc.writeSession(...)and the initsendNextnow run on whichever thread completes the discovery future (a gRPC callback thread) instead of a scheduler thread. That is the point of the change;ReadWriteStreamCallguardssendNext/close/cancelwith aReentrantLock, so the relocation is safe.finally { if (!streamFuture.isDone()) stream.close(); }became awhenCompleteon the partition-id future, registered before the init is sent, so it still fires on both the success path and the sendNext-throws path.doubleStartTestalready asserts the surplus stream is neither started nor closed — andReadWriteStreamCall's constructor starts no RPC, so nothing leaks.onClose/onRetrycan now receive anullstream. When creation itself fails there is no stream to report, soonStreamStop(null, ...)is used.onStreamStoponly passes the argument through toonClose/onRetryand never dereferences it, andWriteSession— the only subclass — ignores the parameter in both. The two abstract methods are documented accordingly.onClose. On that pathclose()finds no registered stream and returnsfalse.WriterImpl.shutdown()already handles exactly that case ("stream will never call onClose") by completingshutdownFutureitself, so no shutdown hangs.Compatibility
Source-incompatible for anyone subclassing
TopicRetryableStreamorWriteStreamFactory:createNewStreamchanged its return type, fromStoCompletableFuture<S>. Both are internal impl classes, and inside the repo the only subclass isWriteSession. Should go into a minor release, not a patch.