Added graceful shutdown for topic streams

[alex268]

No description provided.

[codecov]

Codecov Report

❌ Patch coverage is 98.52941% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 75.49%. Comparing base (d854009) to head (d22442e).

Files with missing lines	Patch %	Lines
...java/tech/ydb/topic/impl/TopicRetryableStream.java	97.36%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #771      +/-   ##
============================================
+ Coverage     75.47%   75.49%   +0.02%     
- Complexity     3755     3757       +2     
============================================
  Files           396      396              
  Lines         16842    16855      +13     
  Branches       1780     1779       -1     
============================================
+ Hits          12711    12725      +14     
- Misses         3515     3517       +2     
+ Partials        616      613       -3     

☔ 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.

[robot-vibe-db]

Review comments on graceful shutdown changes.

[robot-vibe-db]

AI Review Summary

Verdict: ✅ No critical issues found

Critical issues

No critical issues found.

Other findings

• Major | Medium: graceful close now only half-closes the gRPC call and waits for the server to close the stream with no timeout — SyncReaderImpl.shutdown() can block indefinitely on shutdownFuture.join(), and reader metrics gauges are unregistered only when onClose finally fires — topic/src/main/java/tech/ydb/topic/impl/TopicStreamBase.java:71
• Minor | Medium: after close(), state stays non-null in the stream != null branch, so a concurrent fail() reports the application-fail status via onClose and invokes the user's error handler for a user-initiated close (old code suppressed this by nulling the state atomically) — topic/src/main/java/tech/ydb/topic/impl/TopicRetryableStream.java:120
• Minor | Medium: race in State.close() — if the creation future completes between callback registration and the CAS, the new session is started and then half-closed, close() returns false and skips its SUCCESS onStreamStop, so onClose fires later with the transport status — topic/src/main/java/tech/ydb/topic/impl/TopicRetryableStream.java:215
• Minor | Medium: session teardown (preDestroy: decoder stop, partitions closed) now runs before the user's errorHandler in onRetry/onClose, reversing the old order (error handler first, then closeAll) — handlers now see an already-torn-down session — topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java:75
• Minor | Medium: writer graceful-close window — writeQueue is closed only when the server closes the stream; until then SendTask keeps running and queued messages are silently dropped with a warn per message — topic/src/main/java/tech/ydb/topic/impl/TopicStreamBase.java:96
• Minor | Low: ReadSession.preDestroy() is now called repeatedly and concurrently; a late StartPartitionSessionRequest between forEach(close) and clear() adds a partition that is dropped without being closed (onStartPartition has no isClosed guard) — topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java:109
• Minor | Low: test gap — ReadStreamMock(autoClose) always completes with SUCCESS; no coverage of graceful close ending with a non-SUCCESS server status, nor the close()/fail() interleaves — topic/src/test/java/tech/ydb/topic/read/impl/ReadStreamMock.java:66
• Nit | High: typo "nothign" in preDestroy() comment — topic/src/main/java/tech/ydb/topic/impl/TopicStreamBase.java:75
• Nit | Medium: unconditional initFuture.completeExceptionally("Reader closed by user") in shutdown() makes the descriptive "Reader closed with …" message in close(Status) dead in the graceful path; two different messages for the same event depending on timing — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:94
• Nit | Low: send() now merges "before stream is ready" and "after closing" into a single warn message, which makes debugging misuses harder — topic/src/main/java/tech/ydb/topic/impl/TopicRetryableStream.java:105

---

This review was generated automatically. Critical issues require attention; other findings are advisory.
If this comment was useful, please give it a 👍 — it helps us improve the review bot.

[robot-vibe-db]

Full analysis log

Analysis performed by claude, z-ai/glm-5.3-flash.