Per-message processing in topic readers

[alex268]

No description provided.

[codecov]

Codecov Report

❌ Patch coverage is 94.35216% with 17 lines in your changes missing coverage. Please review.
✅ Project coverage is 72.22%. Comparing base (ae5b7db) to head (521dfa8).
⚠️ Report is 4 commits behind head on master.

Files with missing lines	Patch %	Lines
...tech/ydb/topic/read/impl/ReadPartitionSession.java	76.00%	8 Missing and 4 partials ⚠️
...n/java/tech/ydb/topic/read/impl/BufferManager.java	97.56%	1 Missing and 2 partials ⚠️
...ain/java/tech/ydb/topic/read/impl/ReadSession.java	86.66%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master     #695      +/-   ##
============================================
+ Coverage     71.88%   72.22%   +0.34%     
- Complexity     3475     3499      +24     
============================================
  Files           390      392       +2     
  Lines         16206    16373     +167     
  Branches       1698     1726      +28     
============================================
+ Hits          11649    11826     +177     
  Misses         3906     3906              
+ Partials        651      641      -10     

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

AI Review Summary

Verdict: ❌ 1 critical issue(s) found

Critical issues

• Critical | Medium: Missing re-check after isReadingNow release creates a race where decoded messages can be stuck indefinitely — ReadPartitionSession.java:148-150

Other findings

• Major | High: BufferManager.ALL uses Integer.MIN_VALUE/MAX_VALUE instead of Long bounds — buffer leak for offsets > 2^31 — BufferManager.java:24
• Major | Medium: Division by zero in recalcBuffer when all messages have zero-length data but bufferSize > 0 — BufferManager.java:143
• Major | Medium: (int) response.getBytesSize() silently truncates for responses > 2 GB — ReadSession.java:294
• Nit | High: Logger uses ReaderImpl.class instead of BufferManager.class — BufferManager.java:23

---

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, claude-opus-4-6.

[robot-vibe-db]

AI Review Summary

Verdict: ✅ No critical issues found

Critical issues

No critical issues found.

Other findings

• Major | Low: Potential NPE in EncodedMessage.getData() when decompression stream close() throws after successful decode — ReadPartitionDecoder.java:119
• Minor | High: Dead/unreachable code in BufferManager.allocate() — BufferManager.java:97
• Minor | High: ReadPartitionDecoder logger initialized with wrong class (MessageDecoder.class) — ReadPartitionDecoder.java:29
• Minor | Medium: recalcBuffer truncates long buffSize to int in zero-data messages guard — BufferManager.java:139
• Minor | Medium: directDecompressorTest has 10-minute timeout vs 30 seconds for other tests — TopicReadersIntegrationTest.java:289
• Nit | High: Extra leading whitespace on if statement — ReadPartitionSession.java:137

---

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, claude-opus-4-6.

[Copilot]

Pull request overview

Introduces per-message topic-reader decoding with improved flow control and buffer accounting.

Changes:

• Adds per-message decoding and partition queues.
• Adds compressed and decompressed memory management.
• Expands unit and integration coverage.

Reviewed changes

Copilot reviewed 9 out of 9 changed files in this pull request and generated 9 comments.

Show a summary per file

File	Description
TopicReadersIntegrationTest.java	Expands multi-partition reader tests.
MessageDecoderTest.java	Tests decoder scheduling and failures.
BufferManagerTest.java	Tests buffer accounting and concurrency.
ReadSession.java	Integrates new buffer and decoder lifecycle.
ReadPartitionSession.java	Dispatches ready messages individually.
ReadPartitionDecoder.java	Implements partition-level decoding.
MessageImpl.java	Makes message payload handling extensible.
MessageDecoder.java	Schedules per-message decompression.
BufferManager.java	Tracks and replenishes read buffers.

Suppressed comments (1)

topic/src/test/java/tech/ydb/topic/read/impl/BufferManagerTest.java:96

• Correct the typo in this comment: “partitial” should be “partial.”

        // partitial release less than 10% of buffer size - no additional request

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.