feat(inkless:consume): coalesce per-batch buffers and isolate corrupt-batch failures [KC-279]

[jeqo]

Three focused changes to the diskless consume path in FetchCompleter. Together they cut allocation churn on multi-batch fetches, make the fetch response observable, and stop a single bad batch from taking down an entire multi-partition fetch. No change to the wire protocol or to what a healthy fetch returns.

Motivation

Allocation profiling on the diskless consume path showed FetchCompleter.constructRecordsFromFile as the dominant allocator (~38% on one broker, almost entirely byte[]): it allocated one byte[] per batch and copied each extent intersection into it. Separately, investigating that allocation pressure and a DelayedConsolidationFetch OOM repeatedly required allocation profilers because the fetch path emitted no per-fetch byte-volume or per-partition outcome metrics. Finally, while adding those metrics we found that a batch failing integrity validation threw out of servePartition and aborted every partition in the fetch, unlike classic Kafka which isolates read failures per partition.

What changed

1. Coalesce per-batch buffers (refactor)

extractRecords now allocates one partition-level byte[] sized to the sum of all batch ranges and hands each batch a ByteBuffer.slice() of it, instead of one allocation per batch. Slice capacity is bounded to each batch's byte range, so the in-place header mutations (setLastOffset / setMaxTimestamp) stay confined to that batch's region. The median single-batch fetch is unchanged; the multi-batch tail drops from N allocations to 1. Total bytes allocated are unchanged - this is churn/GC-tail relief, not a bytes reduction. A per-batch fallback covers the (practically unreachable) case where the partition total would overflow an int-sized array.

2. Observability metrics (feat)

FetchCompleter now takes a required (non-null) InklessFetchMetrics and records, under both the consumer (io.aiven.inkless.consume:type=InklessFetchMetrics) and consolidation (io.aiven.inkless.consolidation:type=ConsolidationFetchMetrics) JMX groups:

• FetchBytesPerFetchSize (Histogram) - total bytes returned per fetch response; combined with the existing partition/batch fan-out counters this gives the full shape of a fetch (partitions x batches x bytes). Empty/error fetches (0 bytes) are skipped so a caught-up long-poll does not flood the histogram low end.
• PartialPartitionFetchRate (Meter) - partitions whose response was truncated mid-batch-list because a trailing batch was dropped.
• PartitionStorageErrorRate (Meter) - partitions returned with KAFKA_STORAGE_ERROR (missing metadata or missing/incomplete extents).
• PartitionCorruptRecordRate (Meter) - partitions returned with CORRUPT_MESSAGE / INVALID_RECORD because a batch failed integrity validation.

docs/inkless/metrics.rst is regenerated via genInklessMetricsDoc.

3. Isolate corrupt-batch failures to the partition (fix)

A batch that failed integrity validation (CRC / declared size, or a coalesced-run offset mismatch) previously propagated out of servePartition, was wrapped in a FetchException, and failed the entire multi-partition fetch - one bad batch in one partition took down every other partition's valid data. This now matches classic Kafka's contract, where ReplicaManager.readFromLocalLog isolates a read failure to its partition (returning the exception's natural code via Errors.forException) and catches unexpected Throwables per partition:

• constructBatch classifies each batch instead of throwing: missing/non-covering extents -> KAFKA_STORAGE_ERROR; integrity failures -> CorruptRecordException -> CORRUPT_MESSAGE, InvalidRecordException -> INVALID_RECORD. The offset-mismatch / empty-batch guards in createMemoryRecords now raise CorruptRecordException (not IllegalStateException), so a genuine bug is not misclassified as data corruption.
• servePartition serves the valid prefix as a partial response, or the classified error when nothing was extractable. A per-partition safety net catches any unexpected RuntimeException from extraction (oversized batch, corrupt coordinate, truncated extent) and returns a scoped KAFKA_STORAGE_ERROR, so the "never the whole fetch" contract holds for all failure modes, not just the classified ones.
• Dropped corrupt batches are logged at WARN with object key and offset range so operators can locate the bad data.
• Never serves incorrectly-offset data: the corrupt batch is always dropped, never relocated-and-served. On the consumer's next fetch the corrupt batch becomes the first batch and surfaces the classified error directly.

Testing

• FetchCompleterTest and the full io.aiven.inkless.consume.* suite pass; checkstyle passes.
• New/updated tests cover: coalesced-buffer slice isolation on setLastOffset; per-fetch byte metric (including the zero-skip guard); partial/storage-error/corrupt-record metric recording; corrupt-batch isolation returning CORRUPT_MESSAGE for the failing partition while a sibling partition is served normally; valid-prefix service on a corrupt trailing batch; and the coalesced offset-mismatch now returning CORRUPT_MESSAGE per partition instead of throwing.
• Each of the three commits compiles in isolation (the stack is topologically ordered: refactor -> metrics -> isolation).

Notes

• FetchCompleter no longer has an optional/nullable metrics constructor; the metric collaborator is always present.
• Total allocated bytes are unchanged by the coalescing; the win is allocation count and GC-tail pressure on multi-batch (lagging / dense-range) fetches. The predicted young-gen pause improvement is not yet measured in production - the new FetchBytesPerFetchSize metric is the tool to validate it.

[Copilot]

Pull request overview

This PR improves the Inkless (diskless) consume/fetch completion path by reducing allocation churn when assembling multi-batch fetches, adding per-fetch/partition observability metrics, and aligning failure isolation with classic Kafka behavior so that a bad batch only fails its own partition response.

Changes:

• Coalesces per-partition record assembly into a single shared buffer (with per-batch slices) to reduce per-batch allocation churn.
• Adds new JMX metrics for fetch byte volume and per-partition outcomes (partial fetch, storage error, corrupt record).
• Changes batch construction/extraction to classify integrity/incompleteness failures per batch and scope failures to the affected partition (serving a valid prefix where possible).

Reviewed changes

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
storage/inkless/src/main/java/io/aiven/inkless/consume/FetchCompleter.java	Implements buffer coalescing, per-fetch byte metric recording, and per-partition error isolation/classification.
storage/inkless/src/main/java/io/aiven/inkless/consume/InklessFetchMetrics.java	Introduces new histogram/meter metrics and corresponding recording helpers.
storage/inkless/src/main/java/io/aiven/inkless/consume/Reader.java	Wires required InklessFetchMetrics into FetchCompleter.
storage/inkless/src/test/java/io/aiven/inkless/consume/FetchCompleterTest.java	Updates constructor usage and adds/updates tests for metrics and isolated corrupt-batch behavior.
docs/inkless/metrics.rst	Regenerates metrics documentation to include the new Inkless fetch metrics.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated 1 comment.

[Copilot]

Pull request overview

Copilot reviewed 5 out of 5 changed files in this pull request and generated no new comments.

[viktorsomogyi]

I think the commits are sound, just had one question regarding logging.

[viktorsomogyi]

How would a consumer react to a CORRUPT_MESSAGE? Is it retriable by the consumer? Would it flood the logs?

[viktorsomogyi]

Same question as above, can it flood the log?

[jeqo]

good catch, I have fallback to DEBUG and keep the metric as the main signal. PTAL