Added support of RetryConfig topic readers - #722
Open
alex268 wants to merge 2 commits into
Open
Conversation
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #722 +/- ##
============================================
+ Coverage 73.09% 73.30% +0.21%
+ Complexity 3578 3541 -37
============================================
Files 392 391 -1
Lines 16548 16440 -108
Branches 1736 1723 -13
============================================
- Hits 12096 12052 -44
+ Misses 3826 3771 -55
+ Partials 626 617 -9 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
KirillKurdyukov
requested changes
Sep 10, 2026
KirillKurdyukov
left a comment
Contributor
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.
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.
No description provided.