Conversation
Codecov Report❌ Patch coverage is 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. 🚀 New features to boost your workflow:
|
KirillKurdyukov
left a comment
KirillKurdyukov
left a comment
There was a problem hiding this comment.
I found several lifecycle and retry regressions that should be fixed before merge:
ReaderImplnever stores or invokesReaderSettings.getErrorsHandler(). Both retryable and terminal failures are only logged, contradicting the newNEVERdocumentation and regressing the previous retrier behavior.AsyncReaderImpl.shutdown()andSyncReaderImpl.shutdown()ignore a false result fromimpl.close(). Before init or while waiting for a scheduled retry there is no active stream, soonCloseis never called: the async shutdown future stays pending and sync shutdown blocks forever.- A terminal error before the first
InitResponsenever completesinitFuture. WithNEVERorSTANDARD,AsyncReader.init()remains pending andSyncReader.initAndWait()hangs. ReadSession.closeAll()never setsisClosed = 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.- In
ReadSession.onCommitOffset(), encountering one unknown partition executesreturnfrom the whole method. A response may contain several partitions, so acknowledgements and commit futures for later active partitions are skipped. This needscontinue. - Exceptions from user callbacks now go through
impl.fail(CLIENT_INTERNAL_ERROR). The defaultTopicRetryConfig.FOREVERretries that status, so a deterministic callback failure can produce an endless reconnect/redelivery loop; the old implementation stopped the reader. onSessionStartedandonReaderClosedbypass the configured handler executor and serial control-event queue. This changes thread affinity and allowsonReaderClosedto overtake queued partition-closed events.SyncReaderImpl.handleReaderClosed()never closes the default decompressionLazyExecutor. After a compressed message it leaks non-daemon pool threads and can keep the JVM alive.- Terminal stream closure does not signal
waitingCondition, so a thread inreceive()with a long explicit timeout remains blocked until that timeout expires. - 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[bot]](https://avatars.githubusercontent.com/in/3319454?s=60&v=4){kind=small}
There was a problem hiding this comment.
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
shutdownFuturenever completes andshutdown()hangs —topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java:107
Other findings
- Major | High:
onSessionStartednow 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:
onReaderClosedis delivered inline before the queuedonPartitionSessionClosedevents 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.SUCCESSon 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
StopPartitionSessionRequestfor 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
@FixMethodOrderbut still shares mutable topic/consumer state between tests —topic/src/test/java/tech/ydb/topic/YdbTopicsIntegrationTest.java:100 - Minor | Low: duplicate partition start (
putIfAbsenthit) returns without sendingStartPartitionSessionResponse, 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.
|
Analysis performed by claude, z-ai/glm-5.3-flash. |
There was a problem hiding this comment.
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 defaultFOREVERpolicy 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:
LazyExecutorshutdown await raised 100 ms → 5 s, which multiplies the worst-case blocking of every readerclose()caller, including the self-close-from-worker path —topic/src/main/java/tech/ydb/topic/read/impl/LazyExecutor.java:74 - Minor | Medium: duplicate
StartPartitionSessionRequestfor an already-confirmed psid is now silently dropped without sendingStartPartitionSessionResponse(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
StopPartitionSessionRequestfor an unknown partition session is now answered instead of restarting the session (master didcloseDueToError), 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:
handleReaderClosedcatch 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 onClosecomment —topic/src/main/java/tech/ydb/topic/read/impl/SyncReaderImpl.java:97 - Nit | High:
@AfterClass dropTopicno longer asserts drop success, cleanup failures are silent —topic/src/test/java/tech/ydb/topic/YdbTopicsIntegrationTest.java:101 - Nit | High:
TopicRetryConfigclass javadoc still only referencesWriterSettings.Builder#setRetryConfig; now that readers accept aRetryConfig(this PR) the doc/link should mentionReaderSettings.Builder#setRetryConfigtoo —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.
|
Analysis performed by claude, z-ai/glm-5.3-flash. |
c86948d to
92c8a3b
Compare
There was a problem hiding this comment.
AI Review Summary
Verdict: ✅ No critical issues found
Critical issues
- No critical issues found.
Other findings
- Major | High: user
onReaderClosedcallback can be silently dropped or interrupted: it is submitted to the handler executor and thenclose()shuts the LazyExecutor down with only a 100 ms grace beforeshutdownNow()— pending/running user callbacks (including long-runningonMessages) 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 theonReaderClosedevent, 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 (
failSessionfrom user-callback exceptions and failedupdateOffsetsInTransactiontransactions) are silently ignored when no stream is currently active (realStream == nullwhile a reconnect is scheduled): noerrorsHandlernotification 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 useronMessagescallback 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:
updateOffsetsInTransactionno longer requires a live session — called on a closed reader it still sendsUpdateOffsetsInTransactionRequestto the server and registers the tx-failure listener whosefail()becomes a no-op —topic/src/main/java/tech/ydb/topic/read/impl/ReaderImpl.java:151 - Minor | Low: the
impl.isClosed()branch inhandleDataReceivedEventdrops the batch withoutcontrol.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:
buildOffsetRangewas widened topublic staticwithout a need — it is only used internally byReadSession/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.
|
Analysis performed by claude, z-ai/glm-5.3-flash. |
| processor.close(); | ||
| initFuture.completeExceptionally(new RuntimeException("Reader closed with " + status)); | ||
| shutdownFuture.complete(null); |
There was a problem hiding this comment.
[ИИ-ревью] shutdown() завершается раньше завершающих callbacks.
onReaderClosed ставится в очередь, но processor.close() пользовательский executor вообще не ожидает, а стандартный ожидает максимум секунду. Поэтому shutdown().join() возвращается до завершения cleanup в callback: приложение может закрыть ещё используемые ресурсы. Кроме того, ожидание теперь блокирует сам вызов shutdown(), вопреки его неблокирующему контракту. Завершение future следует связать с выполнением завершающих событий, организовав ожидание асинхронно.
There was a problem hiding this comment.
Корректная реализация последовательности событий будет сделана в следубщем PR (это отдельный тикет)
| } else if (message.hasStartPartitionSessionRequest()) { | ||
| StartPartitionSessionEvent event = stream.onStartPartition(message.getStartPartitionSessionRequest()); | ||
| if (event != null) { | ||
| handler.handleStartPartitionSessionRequest(event); |
There was a problem hiding this comment.
[ИИ-ревью] Закрытый поток продолжает порождать события старта партиций.
После локального сбоя, например abort транзакции, старый gRPC-поток закрывается через halfClose, поэтому ещё может доставить сообщения. Прежняя проверка остановленной сессии удалена. Поздний StartPartitionSessionRequest вызывает пользовательский callback, хотя confirm() уже ничего не делает, а соответствующего события закрытия не будет: closeAll() отработал раньше. Требуется отбрасывать сообщения остановленного потока.
There was a problem hiding this comment.
Корректная реализация последовательности событий будет сделана в следубщем PR (это отдельный тикет)
| if (errorHandler != null) { | ||
| try { | ||
| errorHandler.accept(status, null); | ||
| } catch (RuntimeException ex) { | ||
| logger.error("[{}] errorHandler onRetry processing throws exception", debugId, ex); | ||
| } | ||
| } | ||
| stream.closeAll().forEach(ps -> handler.handleClosePartitionSession(ps)); |
There was a problem hiding this comment.
[ИИ-ревью] Остановка reader из обработчика ошибок теряет события закрытия партиций.
errorsHandler вызывается раньше stream.closeAll(). Если обработчик вызывает reader.shutdown(), executor останавливается, и последующие onPartitionSessionClosed молча отбрасываются. В воспроизведении приходит только onReaderClosed; cleanup ресурсов партиций пропускается. Закрытие партиций и постановка их событий должны предшествовать вызову обработчика ошибок. Аналогичный порядок есть в onClose().
There was a problem hiding this comment.
errorHandler - это дополнительный метод раннего информирования об ошибках и да, так и задумано что он должен срабатывать всегда первым. Если пользователь в нем же завершит reader.shutdown() - то логично не получит все последующие сообщения
3 participants
No description provided.