feat(inkless:consolidation): delayed fetcher to honor minBytes/maxWaitMs#685
feat(inkless:consolidation): delayed fetcher to honor minBytes/maxWaitMs#685jeqo wants to merge 3 commits into
Conversation
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.
0b2edd3 to
68b5ecc
Compare
68b5ecc to
82a137c
Compare
a19809f to
f1d9adb
Compare
f1d9adb to
4644e0b
Compare
aefe1a6 to
1b430e5
Compare
f8eddfc to
caa7461
Compare
viktorsomogyi
left a comment
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.
caa7461 to
a604f91
Compare
| val maxOvershootBytes = math.min(logConfig.maxMessageSize, logConfig.segmentSize - 1) | ||
| val segmentSafeFetchSize = math.max(1, logConfig.segmentSize - maxOvershootBytes) | ||
| val partitionFetchSize = math.min(fetchSize, segmentSafeFetchSize) | ||
| if (logConfig.maxMessageSize >= logConfig.segmentSize && unsafeSegmentConfigWarnings.add(topicPartition)) { |
There was a problem hiding this comment.
Do I understand correctly that we can have a situation when maxMessageBytes has been lowered over the time and here we can get an old bigger message that will not fit?
There was a problem hiding this comment.
Agree, this is a valid, pre-existing edge case. A batch committed under a higher max.message.bytes (or a coalesced diskless unit) can exceed the current segment.bytes, and the follower append then rejects it with RecordBatchTooLargeException. Since the batch was already validated and committed at produce time, refusing to replicate it is the wrong outcome.
A few follow-ups worth considering: config-time validation rejecting max.message.bytes > segment.bytes on diskless topics (prevents new oversized batches, but not ones already committed); a per-partition metric for the largest batch/message size; and a stronger fix — an upstream change letting the follower append roll a dedicated oversized segment (what maybeRoll already does for a single batch on the leader path). That last one touches shared upstream code, so it's a deliberate follow-up, not this PR.
For now, adding observability (ERROR log + a per-partition metric) and converting the failure into a retriable per-partition error, so the partition parks and retries with backoff instead of hard-failing. That way bumping segment.bytes self-heals the fetcher without a become-follower/restart.
There was a problem hiding this comment.
Created [KC-345] to track follow-up
…ard-failing When a consolidation fetch serves a block holding a batch larger than the follower's segment.bytes, the follower append throws RecordBatchTooLargeException. That exception was uncaught, so AbstractFetcherThread marked the partition failed and removed it from the fetcher, stalling consolidation until a become-follower event (leader change, reassignment, or broker restart). This happens for a partition holding a batch larger than the current segment.bytes, e.g. max.message.bytes lowered below segment.bytes over time or a coalesced diskless unit. The batch was already validated and committed at produce time, so refusing to replicate it is the wrong outcome. Catch RecordBatchTooLargeException in ConsolidationFetcherThread and rethrow as ConsolidationSegmentOverflowException (extends InvalidRecordException), which AbstractFetcherThread treats as a soft, retriable per-partition error: the partition stays parked at the same offset and retries with backoff, so raising segment.bytes resumes consolidation without operator intervention on the fetcher. Add a per-partition ConsolidationOversizedBatch counter plus an ERROR log to make the condition observable.
| * Trim a fetched block so the follower's single append fits `segment.bytes`, restoring the classic | ||
| * invariant `returned <= maxBytes + one physical batch (<= max.message.bytes) <= segment.bytes`. |
There was a problem hiding this comment.
I think we are not following this invariant here because the last overshoot message can be bigger than max.message.bytes and can exceed the segment.bytes
| // bounded per partition by diskless.consolidation.fetch.max.bytes and in aggregate by | ||
| // diskless.consolidation.fetch.response.max.bytes. |
There was a problem hiding this comment.
But is the aggregate boundary respected? I think the first coordinate of every partition will be fetched unconditionally and that can overflow the boundary.
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.