feat(inkless:consolidation): delayed fetcher to honor minBytes/maxWaitMs#685
Open
jeqo wants to merge 2 commits into
Open
feat(inkless:consolidation): delayed fetcher to honor minBytes/maxWaitMs#685jeqo wants to merge 2 commits into
jeqo wants to merge 2 commits into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR updates the inkless consolidation fetch path to honor diskless.consolidation.fetch.{min.bytes,max.wait.ms} by introducing a long-poll style delayed operation, reducing control-plane (PG) load when partitions are idle, and adjusts the default fetch sizing knobs accordingly.
Changes:
- Add
DelayedConsolidationFetchand a dedicateddelayedConsolidationFetchPurgatory, then use it fromDisklessLeaderEndPoint.fetchto park idle fetches untilminBytesis reached ormaxWaitMsexpires. - Increase consolidation fetch defaults (
max.bytes,response.max.bytes,min.bytes,max.wait.ms) and update the corresponding config assertions. - Add unit tests covering delayed-fetch completion/parking/expiration/error behavior.
Reviewed changes
Copilot reviewed 7 out of 7 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| server-common/src/main/java/org/apache/kafka/server/config/ServerConfigs.java | Raises default consolidation fetch sizing/wait thresholds now that they’re honored. |
| docs/inkless/configs.rst | Updates config docs text (currently introduces two encoding artifacts flagged in comments). |
| core/src/test/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPointTest.scala | Updates expected defaults for consolidation fetch config values. |
| core/src/test/scala/io/aiven/inkless/consolidation/DelayedConsolidationFetchTest.scala | Adds new test coverage for delayed consolidation fetch semantics. |
| core/src/main/scala/kafka/server/ReplicaManager.scala | Adds delayedConsolidationFetchPurgatory, documents findDisklessBatches behavior, and shuts down the new purgatory. |
| core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala | Routes consolidation fetch through the new delayed-fetch purgatory path. |
| core/src/main/scala/io/aiven/inkless/consolidation/DelayedConsolidationFetch.scala | Implements delayed operation: single metadata probe, park until timeout, then perform real fetch and return results/errors. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
jeqo
force-pushed
the
jeqo/kc-278-delayed-consolidation-fetch
branch
from
July 6, 2026 11:48
0b2edd3 to
68b5ecc
Compare
jeqo
changed the title
jeqo
force-pushed
the
jeqo/kc-278-delayed-consolidation-fetch
branch
from
July 6, 2026 14:10
68b5ecc to
82a137c
Compare
jeqo
force-pushed
the
jeqo/kc-278-delayed-consolidation-fetch
branch
3 times, most recently
from
July 6, 2026 17:24
a19809f to
f1d9adb
Compare
jeqo
force-pushed
the
jeqo/kc-278-delayed-consolidation-fetch
branch
from
July 6, 2026 17:36
f1d9adb to
4644e0b
Compare
jeqo
marked this pull request as ready for review
July 6, 2026 18:37
jeqo
force-pushed
the
jeqo/kc-278-delayed-consolidation-fetch
branch
2 times, most recently
from
July 7, 2026 07:46
aefe1a6 to
1b430e5
Compare
jeqo
force-pushed
the
jeqo/kc-278-delayed-consolidation-fetch
branch
3 times, most recently
from
July 8, 2026 09:46
f8eddfc to
caa7461
Compare
viktorsomogyi
previously approved these changes
Jul 14, 2026
viktorsomogyi
left a comment
viktorsomogyi
left a comment
Contributor
There was a problem hiding this comment.
All good from my side, but @AnatolyPopov mentioned yesterday that he'd like to take a look, so I'm not merging yet to give him a chance.
…s per batch buildFetch reserves per-partition maxBytes = segment.bytes - max.message.bytes so the follower's single append cannot exceed the destination segment. That headroom assumes find_batches overshoots by at most one physical batch (<= max.message.bytes), which is false under commit_file_v2: find_batches limits at coordinate granularity and always admits the coordinate that crosses maxBytes, and a coalesced coordinate can contain many byte-contiguous physical batches whose total exceeds max.message.bytes (up to produce.buffer.max.bytes). The admitted coordinate, or even the first always-admitted one, can push the block past a small segment.bytes and fail the follower append with RecordBatchTooLargeException. This is reachable when segment.bytes is tuned small (e.g. for eager tiering). Enforce the reserved budget at physical-batch granularity in fetch() via clampRecordsToSegment, restoring the classic invariant returned <= maxBytes + one physical batch (<= max.message.bytes) <= segment.bytes. Batches with lastOffset < fetchOffset are dropped so a re-fetch into a coalesced coordinate that a prior fetch truncated mid-run does not re-serve the already-appended prefix; at least one batch is always emitted for progress. When nothing is trimmed the original records are returned unchanged to preserve the ConcatenatedRecords zero-copy path.
jeqo
force-pushed
the
jeqo/kc-278-delayed-consolidation-fetch
branch
from
July 17, 2026 10:06
caa7461 to
a604f91
Compare
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.
The consolidation fetcher previously called Reader.fetch() synchronously on every doWork() iteration regardless of whether data was available. Idle partitions therefore queried the control plane on every loop iteration: it returned empty and the loop ran again. Existing configs diskless.consolidation.fetch.{min.bytes,max.wait.ms} were passed into FetchParams but Reader.fetch() ignored them.
Wrap the fetcher path in DisklessLeaderEndPoint.fetch with a DelayedConsolidationFetch that parks in a new
delayedConsolidationFetchPurgatory until the initial probe sees minBytes of accumulated diskless bytes or maxWaitMs expires. tryComplete runs one cheap findDisklessBatches probe (gated by a CAS so tryCompleteElseWatch's before/after double-check probes only once); onComplete runs the full FetchHandler.handle once. This purgatory has no wake-up sites, so parked ops complete via the maxWaitMs timeout, keeping control-plane pressure bounded regardless of loop rate.
Set the purgatory purgeInterval to 0 (matches delayedRemoteFetchPurgatory) so completed ops release their captured response references immediately for GC. Each completed op pins a Map[TopicIdPartition, FetchPartitionData] of fetched records; without aggressive purging, watch lists accumulate up to fetchPurgatoryPurgeIntervalRequests completed ops and exhaust the heap.
Raise the consolidation fetch defaults now that min.bytes/max.wait.ms are honored: max.bytes 1MB to 10MB (per partition), response.max.bytes 10MB to 64MB (total), min.bytes 1 to 8MB, max.wait.ms 500 to 1000.
See commits for details.