Added support of RetryConfig topic readers

[alex268]

No description provided.

[codecov]

Codecov Report

❌ Patch coverage is 86.98225% with 66 lines in your changes missing coverage. Please review.
✅ Project coverage is 74.38%. Comparing base (d77889d) to head (c8d73e4).

Files with missing lines	Patch %	Lines
...main/java/tech/ydb/topic/read/impl/ReaderImpl.java	87.91%	8 Missing and 10 partials ⚠️
...java/tech/ydb/topic/read/impl/AsyncReaderImpl.java	80.72%	16 Missing ⚠️
...ain/java/tech/ydb/topic/read/impl/ReadSession.java	88.28%	7 Missing and 8 partials ⚠️
.../java/tech/ydb/topic/read/impl/SyncReaderImpl.java	84.61%	6 Missing and 6 partials ⚠️
...in/java/tech/ydb/topic/read/impl/LazyExecutor.java	0.00%	1 Missing and 1 partial ⚠️
...tech/ydb/topic/read/impl/MessageCommitterImpl.java	88.23%	2 Missing ⚠️
...main/java/tech/ydb/topic/impl/TopicStreamBase.java	88.88%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #722      +/-   ##
============================================
+ Coverage     73.18%   74.38%   +1.19%     
- Complexity     3589     3615      +26     
============================================
  Files           392      391       -1     
  Lines         16531    16466      -65     
  Branches       1736     1733       -3     
============================================
+ Hits          12099    12249     +150     
+ Misses         3811     3604     -207     
+ Partials        621      613       -8     

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

[KirillKurdyukov]

I found several lifecycle and retry regressions that should be fixed before merge:

1. ReaderImpl never stores or invokes ReaderSettings.getErrorsHandler(). Both retryable and terminal failures are only logged, contradicting the new NEVER documentation and regressing the previous retrier behavior.
2. AsyncReaderImpl.shutdown() and SyncReaderImpl.shutdown() ignore a false result from impl.close(). Before init or while waiting for a scheduled retry there is no active stream, so onClose is never called: the async shutdown future stays pending and sync shutdown blocks forever.
3. A terminal error before the first InitResponse never completes initFuture. With NEVER or STANDARD, AsyncReader.init() remains pending and SyncReader.initAndWait() hangs.
4. ReadSession.closeAll() never sets isClosed = true. Committing an already delivered message after reconnect creates a future, reports the send as accepted, then silently drops the request because the underlying stream is closed. That future can never complete.
5. In ReadSession.onCommitOffset(), encountering one unknown partition executes return from the whole method. A response may contain several partitions, so acknowledgements and commit futures for later active partitions are skipped. This needs continue.
6. Exceptions from user callbacks now go through impl.fail(CLIENT_INTERNAL_ERROR). The default TopicRetryConfig.FOREVER retries that status, so a deterministic callback failure can produce an endless reconnect/redelivery loop; the old implementation stopped the reader.
7. onSessionStarted and onReaderClosed bypass the configured handler executor and serial control-event queue. This changes thread affinity and allows onReaderClosed to overtake queued partition-closed events.
8. SyncReaderImpl.handleReaderClosed() never closes the default decompression LazyExecutor. After a compressed message it leaks non-daemon pool threads and can keep the JVM alive.
9. Terminal stream closure does not signal waitingCondition, so a thread in receive() with a long explicit timeout remains blocked until that timeout expires.
10. A graceful stop request for an unknown partition is now only logged and ignored. The previous implementation restarted the stream; the new behavior leaves the server waiting for a response and the protocol state inconsistent.

The existing test suite passes, but this PR adds no retry/lifecycle tests. Codecov patch coverage is 66.22% with 152 changed lines uncovered. Please add targeted tests for terminal-before-init, shutdown-during-backoff, stale-message commit, multi-partition commit acknowledgements, callback failure, and executor cleanup.

[robot-vibe-db]

AI Review Summary

Verdict: ❌ 1 critical issue(s) found

Critical issues

• Critical | High: errors handler is invoked twice on reader close and the second invocation is unguarded; a throwing handler aborts the whole close sequence so shutdownFuture never completes and shutdown() hangs — topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java:107

Other findings

• Major | High: onSessionStarted now runs inline on the gRPC stream thread instead of the control executor; a blocking callback stalls the whole read stream and event ordering across reconnects is broken — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:126
• Major | High: onReaderClosed is delivered inline before the queued onPartitionSessionClosed events and the executor shutdown can silently drop those queued events — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:135
• Major | High: the errors handler is now invoked with Status.SUCCESS on graceful shutdown, inconsistent with the writer implementation and with the previous reader behavior — topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java:101
• Major | Medium: an exception from a user callback now restarts the reader indefinitely under the default retry policy instead of stopping it (and is silently ignored when no stream is active) — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:118
• Major | Medium: init()/initAndWait() complete successfully even when the reader closed with an error; the old code failed the init future and the writer still does — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:50
• Minor | High: dead handleReaderClosed() helper with different (executor-based) semantics than the active inline path — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:88
• Minor | Medium: user callbacks are interrupted on shutdown (100 ms grace period, then shutdownNow()), while the old code shut the pool down gracefully — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:110
• Minor | Medium: a graceful StopPartitionSessionRequest for an unknown partition session is now silently acked instead of restarting the session, hiding a possible SDK/server desync — topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java:172
• Minor | Medium: the integration suite dropped @FixMethodOrder but still shares mutable topic/consumer state between tests — topic/src/test/java/tech/ydb/topic/YdbTopicsIntegrationTest.java:100
• Minor | Low: duplicate partition start (putIfAbsent hit) returns without sending StartPartitionSessionResponse, leaving the server-side session unconfirmed — topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java:288

---

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.

[robot-vibe-db]

AI Review Summary

Verdict: ✅ No critical issues found

Critical issues

No critical issues found.

Other findings

• Major | Medium: exception in a user callback now triggers impl.fail() + reconnect instead of the old terminal stop — with the default FOREVER policy a poison message causes an infinite crash/reconnect redelivery loop — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:107
• Major | Medium: terminal close blocks the gRPC callback thread up to ~10 s (decompressor.close() + processor.close(), each awaiting 5 s; master closed executors without awaiting) — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:135
• Minor | Medium: LazyExecutor shutdown await raised 100 ms → 5 s, which multiplies the worst-case blocking of every reader close() caller, including the self-close-from-worker path — topic/src/main/java/tech/ydb/topic/read/impl/LazyExecutor.java:74
• Minor | Medium: duplicate StartPartitionSessionRequest for an already-confirmed psid is now silently dropped without sending StartPartitionSessionResponse (master always responded), which can stall the partition server-side — topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java:286
• Minor | Medium: test gap — async reader retry/close/callback-failure paths have no unit coverage (only the sync reader is unit-tested; async paths are only in skipped integration tests) — topic/src/test/java/tech/ydb/topic/read/impl/ReaderImplTest.java:29
• Minor | Low: graceful StopPartitionSessionRequest for an unknown partition session is now answered instead of restarting the session (master did closeDueToError), masking possible client/server state desync — topic/src/main/java/tech/ydb/topic/read/impl/ReadSession.java:172
• Minor | Low: sync reader no longer releases the commit range of skipped inactive messages (next.release() removed), so buffer bytes from a stopped partition are reclaimed only at full partition release — topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java:179
• Nit | High: handleReaderClosed catch block passes "onSessionStarted" as the callback name (copy-paste) — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:130
• Nit | High: duplicated // implicit closing because stream will never call onClose comment — topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java:97
• Nit | High: @AfterClass dropTopic no longer asserts drop success, cleanup failures are silent — topic/src/test/java/tech/ydb/topic/YdbTopicsIntegrationTest.java:101
• Nit | High: TopicRetryConfig class javadoc still only references WriterSettings.Builder#setRetryConfig; now that readers accept a RetryConfig (this PR) the doc/link should mention ReaderSettings.Builder#setRetryConfig too — topic/src/main/java/tech/ydb/topic/settings/TopicRetryConfig.java:16 (file not part of this diff, noted here only)

---

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.

[robot-vibe-db]

AI Review Summary

Verdict: ✅ No critical issues found

Critical issues

• No critical issues found.

Other findings

• Major | High: user onReaderClosed callback can be silently dropped or interrupted: it is submitted to the handler executor and then close() shuts the LazyExecutor down with only a 100 ms grace before shutdownNow() — pending/running user callbacks (including long-running onMessages) prevent it from running — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:136
• Major | Medium: shutdown() while the reader is between reconnection attempts completes futures directly and never fires the onReaderClosed event, even though the reader was actively running (partitions started, messages delivered) — whether the event is delivered now depends on reconnect timing — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:90
• Major | Medium: app-side fatal errors (failSession from user-callback exceptions and failed updateOffsetsInTransaction transactions) are silently ignored when no stream is currently active (realStream == null while a reconnect is scheduled): no errorsHandler notification and no shutdown, the reader just keeps reconnecting and re-delivering — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:107
• Minor | Medium: with the default TopicRetryConfig.FOREVER, an exception thrown by a user onMessages callback now triggers an endless reconnect/re-delivery loop (the old SDK permanently stopped the reader), repeatedly replaying the same failing batch — topic/src/main/java/tech/ydb/topic/read/impl/AsyncReaderImpl.java:150
• Minor | Medium: updateOffsetsInTransaction no longer requires a live session — called on a closed reader it still sends UpdateOffsetsInTransactionRequest to the server and registers the tx-failure listener whose fail() becomes a no-op — topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java:151
• Minor | Low: the impl.isClosed() branch in handleDataReceivedEvent drops the batch without control.confirmRangeProcessed(...), unlike the empty-batch branch, so the range/buffers are never released on that path — topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java:228
• Nit | High: buildOffsetRange was widened to public static without a need — it is only used internally by ReadSession/ReaderImpl — topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java:243

---

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.