Conversation
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>
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #726 +/- ##
============================================
- Coverage 74.54% 74.47% -0.08%
- Complexity 3637 3641 +4
============================================
Files 392 391 -1
Lines 16493 16518 +25
Branches 1738 1741 +3
============================================
+ Hits 12294 12301 +7
- Misses 3591 3597 +6
- Partials 608 620 +12 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
![robot-vibe-db[bot]](https://avatars.githubusercontent.com/in/3319454?s=60&v=4){kind=small}
| String streamId = realStreamId.getAndSet(null); | ||
| if (streamId != null) { | ||
| logger.warn("[{}] failed by application-side error {}", streamId, status); | ||
| S local = realStream; |
There was a problem hiding this comment.
Severity: Major
Confidence: Medium
fail() (and close()) now tear down the stream in two independent steps: realStreamId.getAndSet(null) followed by a separate volatile read of realStream. If a pending createNewStream future completes in between (possible when start() was called twice, or a user start() races a retry reconnect), tryStartStream wins the CAS(null, id2) and installs a new stream, and this method then reads local = realStream and closes the new stream — while the stream this call intended to stop is abandoned unmanaged: its own stop callback loses the realStreamId.compareAndSet(streamID, null) in startStream, so its termination is never processed and it keeps delivering messages (e.g. duplicate data for a reader) until the server closes it. The old code was immune because it swapped the stream with a single atomic realStream.getAndSet(null).
Consider keeping the id and the stream in one atomic reference (e.g. AtomicReference of an (id, stream) entry) so they are captured and cleared together, and adding a test for fail()/close() interleaved with an in-flight creation.
| } | ||
|
|
||
| private void tryStartStream(String streamID, Result<S> result, Throwable th) { | ||
| if (isClosed) { |
There was a problem hiding this comment.
Severity: Major
Confidence: High
Both early-return paths of tryStartStream (this isClosed skip and the "double start" CAS failure below) discard an already-created stream without ever starting or closing it, leaking the underlying gRPC call that the factories create eagerly (rpc.writeSession(...) in WriteStreamFactory/WriteStreamDirectFactory, rpc.readSession(...) in ReaderImpl). The isClosed branch is newly reachable and easy to hit: close() while a stream is being created — e.g. writer.shutdown() during a direct-write initial connect or reconnect, where the probe stream plus describeTopic (1-minute deadline) can be in flight for a long time — guarantees the leak. closeWhileStreamIsCreatingTest codifies that the stream "must not be started", but never asserts it is closed either.
Suggested fix: on both skip paths, close the created stream when the result is successful, e.g. if (result != null && result.isSuccess()) result.getValue().close(); — and consider propagating cancellation into the creation future so the probe/describe work is also abandoned on close().
|
|
||
| String streamID = debugId + '.' + streamCount.incrementAndGet(); | ||
| S stream = createNewStream(streamID); | ||
| createNewStream(streamID).whenComplete((result, th) -> { |
There was a problem hiding this comment.
Severity: Minor
Confidence: Low
Anything thrown inside tryStartStream — including exceptions from user callbacks reached via onStreamStop → onRetry/onClose — is captured by the future returned from whenComplete, which nobody observes, so retry/stop handling silently dies mid-flight. Previously the first start() ran this logic on the caller thread, so a synchronous failure at least surfaced to the caller.
Consider wrapping the callback body in try/catch with a logger.error, so a failure here is at least visible in the logs.
| } | ||
|
|
||
| if (result == null) { | ||
| logger.warn("[{}] cannot create stream with exeption", streamID, th); |
There was a problem hiding this comment.
Severity: Nit
Confidence: High
Typo in the log message: "exeption" → "exception".
| * @param id identifier of the new stream for logging | ||
| * @return future with the new stream | ||
| */ | ||
| public CompletableFuture<Result<WriteSession.Stream>> createNewStream(String id) { |
There was a problem hiding this comment.
Severity: Minor
Confidence: High
This is a source-incompatible signature change on a public extension point: WriteStreamFactory is user-extensible (a custom instance can be passed to the writer constructor), so custom subclasses overriding createNewStream will stop compiling. The same applies to subclasses of TopicRetryableStream whose onRetry/onClose now receive a @Nullable stream. Since the class lives in an impl package this may be acceptable, but it deserves a mention in the release notes; otherwise consider keeping a delegating overload for the old signature.
| )); | ||
| } | ||
|
|
||
| private WriteSession.Stream buildDirectStream(String id, long partitionId, YdbTopic.PartitionLocation location) { |
There was a problem hiding this comment.
Severity: Nit
Confidence: High
The parameter partitionId shadows the partitionId field of WriteStreamFactory. It is used consistently inside this method, but the shadowing makes it easy to confuse the resolved partition id with the configured one — consider renaming the parameter (e.g. resolvedPartitionId).
AI Review SummaryVerdict: ✅ No critical issues found Critical issuesNo critical issues found. Other findings
The core change itself (non-blocking stream creation, This review was generated automatically. Critical issues require attention; other findings are advisory. |
|
Analysis performed by claude, z-ai/glm-5.3-flash. |
1 participant
No description provided.