8000 fix(inkless:control_plane): make the diskless fetch budget fair across partitions [KC-407] by jeqo · Pull Request #750 · aiven/inkless · GitHub
[go: up one dir, main page]
Skip to content

fix(inkless:control_plane): make the diskless fetch budget fair across partitions [KC-407] - #750

Merged
AnatolyPopov merged 5 commits into
mainfrom
jeqo/fetch-fairness
Aug 21, 2026
Merged

fix(inkless:control_plane): make the diskless fetch budget fair across partitions [KC-407]#750
AnatolyPopov merged 5 commits into
mainfrom
jeqo/fetch-fairness

Conversation

@jeqo
@jeqo jeqo commented Aug 12, 2026 •
edited
Loading
Copy link
Copy Markdown
Contributor

find_batches guarantees every partition in a fetch at least one batch. Upstream grants that allowance once per request and returns zero bytes past the shared fetch.max.bytes, and that zero-byte response is the only input to Kafka's fetch fairness — both readers of this path demote a partition only when it returned bytes. The per-partition floor therefore makes every partition look served, the rotation never advances, and lagging partitions sit at one batch per round while the head drains: catch-up serializes instead of running concurrently.

Restoring the upstream semantics fixes the consumer path and the consolidation fetcher together — same function, same rotation rule — with no consolidation-specific change.

Commits

  • preserve fetch-session order into the diskless control plane — fetchInfos.toMap discarded the rotated order; above 4 entries immutable.Map is a HashMap whose iteration order is a pure function of the key set, so it was identical on every fetch.
  • make find_batches minOneMessage request-level — one predicate in find_batches_v2 (shipped as migration V27).
  • pin the empty-response contract the fetch rotation needs — test-only guard on the consolidation path.

Not reorderable: commit 2 without commit 1 is worse than the bug (request order stays a fixed HashMap order, so the tail gets zero instead of one, permanently). See the commit messages for the reasoning behind each step.

The SQL change

The change to find_batches_v2 ships as a CREATE OR REPLACE FUNCTION, so the raw diff shows the whole function as new. The function body differs from V19 by one predicate (the surrounding added lines are explanatory comments):

-            IF l_partition_batch_count = 0
+            IF l_global_bytes = 0
                 OR (l_partition_bytes < l_request.max_partition_fetch_bytes
                     AND l_global_bytes < fetch_max_bytes)
diff <(sed -n '/^CREATE OR REPLACE/,$p' storage/inkless/src/main/resources/db/migration/V19__Find_batches_bounded_scan.sql) \
     <(sed -n '/^CREATE OR REPLACE/,$p' storage/inkless/src/main/resources/db/migration/V27__Find_batches_request_level_min_one_message.sql)

The bounded O(result) scan and its index-scan plan dependency, error handling, request ordering and max_batches_per_partition are untouched.

InMemoryControlPlane (selectable for local and system-test deployments) previously applied fetch.max.bytes per partition and ignored both the per-partition byte cap and the batch cap; it now enforces all three so the same expectation can be asserted against both control planes. Tests that relied on the in-memory plane ignoring the per-partition cap see fewer batches.

Behaviour change

A partition with data available can now return an empty response when the request budget is spent. FetchCompleter maps an empty batch list to MemoryRecords.EMPTY with Errors.NONE and the real high watermark — the response a caught-up partition already gets — so lag stays visible.

That empty path had a latent gap this change surfaces: it dropped lastStableOffset (passing OptionalLong.empty() where the served branch carries the high watermark), which reaches a READ_COMMITTED consumer as INVALID_LAST_STABLE_OFFSET and skips SubscriptionState.updateLastStableOffset, staling currentLag/endOffsets. It was harmless while only a caught-up partition reached the branch; under request-level budgeting a lagging partition reaches it on every round it waits its turn. The empty success path now carries the high watermark as the LSO, and the error path keeps it unset.

Response size is clamped again as a consequence: overshoot drops from one batch per partition to one per request.

Batches served per partition, before → after: consumer profile (4 partitions, 4 rounds) [32, 4, 4, 4] → [8, 8, 8, 8]; consolidation budget ratio (10 partitions, 10 rounds) [50 x5, 10 x5] → 25 each, at 300 → 250 total batches fetched — so this is cheaper, not a fairness/throughput tradeoff.

Testing

Cross-request budget fairness is pinned end to end: FindBatchesFairnessTest replays each caller's move-to-end rotation against the real SQL and asserts every partition advances equally over many rounds (consumer and consolidation profiles); FindBatchesJobTest and AbstractControlPlaneTest assert that partitions past the budget return no batches, the latter against both control planes. FetchCompleterTest covers the empty-but-lagging response carrying the LSO; ConsolidationFetcherThreadTest and ReplicaManagerInklessTest guard the empty-response and order-preservation plumbing.

Five parameterised rows across three existing tests asserted the floor; they are relocated into findServesOnlyTheFirstPartitionWhenTheRequestBudgetIsSpent with the second partition now expected empty. No existing expectation was adjusted in place.

Follow-up (not in this PR)

fetchParamsWithNewMaxBytes is an identity function — Math.max((maxBytes * percentage).toInt, maxBytes) with every caller passing percentage <= 1 — so both the classic and diskless legs receive the full fetch.max.bytes, contradicting its docstring. Independent defect, deliberately deferred: the obvious fix splits by partition count, which can cut diskless catch-up N-fold on mixed requests and still would not bound consolidating fetches. Filed as a gap with measurements needed first.

The order invariant that fetchDisklessMessages depends on is carried by javadoc and tests, not by a type: Seq[(TopicIdPartition, PartitionData)] cannot express that its order is load-bearing. A per-caller order-propagation test would guard the producers directly; a type-level change would not (a Seq/LinkedHashMap is ordered but cannot encode that the order is the rotation order).

@jeqo
jeqo requested a balanced review from Copilot August 12, 2026 22:26
Copilot AI reviewed Aug 12, 2026
View reviewed changes
Copilot AI left a comment
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

Makes diskless fetch budgeting fair by preserving request order and applying the minimum-one-message allowance once per request.

Changes:

  • Preserves fetch-session order through LinkedHashMap.
  • Updates find_batches_v2 to enforce request-level budgeting.
  • Adds fairness, empty-response, and ordering tests; regenerates jOOQ schema metadata.

Reviewed changes

Copilot reviewed 6 out of 124 changed files in this pull request and generated no comments.

Show a summary per file
File Description
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/FindBatchesJobTest.java Tests request-level budget behavior.
storage/inkless/src/test/java/io/aiven/inkless/control_plane/postgres/FindBatchesFairnessTest.java Tests fairness across fetch rounds.
storage/inkless/src/main/resources/db/migration/V25__Find_batches_request_level_min_one_message.sql Applies request-level minimum-one-message semantics.
core/src/main/scala/kafka/server/ReplicaManager.scala Preserves fetch-session partition order.
core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala Verifies ordered handoff.
core/src/test/scala/io/aiven/inkless/consolidation/ConsolidationFetcherThreadTest.scala Verifies empty consolidation responses.
storage/inkless/src/main/jooq/org/jooq/generated/UDTs.java Updates generated schema metadata to v25.
storage/inkless/src/main/jooq/org/jooq/generated/udt/RepairDisklessLogResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/RepairDisklessLogRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/PruneBatchesBelowHighestTieredOffsetResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/PruneBatchesBelowHighestTieredOffsetRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/ListOffsetsResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/ListOffsetsRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/InitDisklessLogProducerStateV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/FindBatchesResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/FindBatchesRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/EnforceRetentionResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/EnforceRetentionRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/DeleteRecordsResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/DeleteRecordsRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/CommitBatchResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/CommitBatchRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/BatchMetadataV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/BatchInfoV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/AdvanceCrossTierLogStartResponseV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/AdvanceCrossTierLogStartRequestV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/RepairDisklessLogResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/RepairDisklessLogRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/PruneBatchesBelowHighestTieredOffsetResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/PruneBatchesBelowHighestTieredOffsetRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/ListOffsetsResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/ListOffsetsRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/InitDisklessLogProducerStateV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/FindBatchesResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/FindBatchesRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/EnforceRetentionResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/EnforceRetentionRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/DeleteRecordsResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/DeleteRecordsRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/CommitBatchResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/CommitBatchRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/BatchMetadataV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/BatchInfoV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/AdvanceCrossTierLogStartResponseV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/records/AdvanceCrossTierLogStartRequestV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/RepairDisklessLogResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/RepairDisklessLogRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/PruneBatchesBelowHighestTieredOffsetResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/PruneBatchesBelowHighestTieredOffsetRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/ListOffsetsResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/ListOffsetsRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/InitDisklessLogProducerStateV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/FindBatchesResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/FindBatchesRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/EnforceRetentionResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/EnforceRetentionRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/DeleteRecordsResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/DeleteRecordsRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/CommitBatchResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/CommitBatchRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/BatchMetadataV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/BatchInfoV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/AdvanceCrossTierLogStartResponseV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/udt/paths/AdvanceCrossTierLogStartRequestV1Path.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/RepairDisklessLogV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/PruneBatchesBelowHighestTieredOffsetV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/ProducerState.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/Logs.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/ListOffsetsV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/InitDisklessLogV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/FindBatchesV2.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/FindBatchesV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/Files.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/EnforceRetentionV2.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/EnforceRetentionV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/DeleteRecordsV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/CommitFileV2.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/CommitFileV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/Batches.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/AdvanceCrossTierLogStartV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/RepairDisklessLogV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/PruneBatchesBelowHighestTieredOffsetV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/ProducerStateRecord.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/LogsRecord.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/ListOffsetsV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/InitDisklessLogV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FindBatchesV2Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FindBatchesV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/FilesRecord.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/EnforceRetentionV2Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/EnforceRetentionV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/DeleteRecordsV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/CommitFileV2Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/CommitFileV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/BatchesRecord.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/tables/records/AdvanceCrossTierLogStartV1Record.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Tables.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Routines.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Keys.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Indexes.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/Domains.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/DefaultSchema.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/routines/MarkFileToDeleteV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/routines/FlushCommitRunV2.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteTopicV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteFilesV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/routines/DeleteBatchV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/routines/BatchTimestamp.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/PruneBatchesBelowHighestTieredOffsetErrorV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/ListOffsetsResponseErrorV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/InitDisklessLogErrorV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/FindBatchesResponseErrorV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/FileStateT.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/FileReasonT.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/EnforceRetentionResponseErrorV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/DeleteRecordsResponseErrorV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/CommitBatchResponseErrorV1.java Updates schema metadata.
storage/inkless/src/main/jooq/org/jooq/generated/enums/AdvanceCrossTierLogStartResponseErrorV1.java Updates schema metadata.

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

@jeqo
jeqo marked this pull request as ready for review August 13, 2026 07:45
@jeqo
jeqo requested a review from AnatolyPopov August 13, 2026 07:46
@jeqo jeqo changed the title fix(inkless:consume): Make the diskless fetch budget fair across partitions [KC-407] fix(inkless:control_plane): make the diskless fetch budget fair across partitions [KC-407] Aug 17, 2026
@jeqo
jeqo force-pushed the jeqo/fetch-fairness branch 2 times, most recently from f659ddc to 2f4ca2c Compare August 20, 2026 10:26
@jeqo
jeqo requested a balanced review from Copilot August 20, 2026 10:38
Copilot AI reviewed Aug 20, 2026
View reviewed changes
Copilot AI left a comment
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pull request overview

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

@jeqo
jeqo force-pushed the jeqo/fetch-fairness branch from 2f4ca2c to 0b90462 Compare August 20, 2026 10:45
@jeqo
jeqo requested a review from AnatolyPopov August 20, 2026 10:45
@jeqo
jeqo dismissed AnatolyPopov’s stale review August 21, 2026 07:08

Suggestions applied

jeqo added 3 commits August 21, 2026 13:36
@jeqo
fix(inkless): preserve fetch-session order into the diskless control …
f6e9f0c 
…plane [KC-407]

find_batches_v2 spends a fetch's shared fetch.max.bytes budget in request-array
order, and that array is built by iterating the map handed to the diskless fetch
handler. The broker had already computed a fair order: FetchSession moves every
partition that returned bytes to the back of session.partitionMap, so a starved
partition drifts to the front and is served first on the next fetch. That order
reached fetchDisklessMessages intact, as an ordered Seq.

fetchInfos.toMap discarded it. scala.collection.immutable.Map has dedicated
Map1..Map4 classes that preserve insertion order, but at five or more entries it
is a HashMap whose iteration order ignores insertion and is a pure function of
the key set -- so it is identical on every subsequent fetch for a stable
partition assignment. Hash order thereby replaced the rotation with a fixed
priority ranking keyed on partition identity.

This does not by itself make the allocation fair. The rotation it restores stays
inert while find_batches guarantees every partition at least one batch, because
FetchSession only demotes partitions that returned bytes and under that floor
they all do. This is a prerequisite for that fix, not a substitute for it.

The new test uses eight partitions deliberately; four or fewer pass even against
the discarded order.
@jeqo
fix(inkless:control_plane): make find_batches minOneMessage request-l…
923c32f 
…evel [KC-407]

find_batches_v2 admitted the first batch of every partition unconditionally --
"always >=1 batch per partition", inherited from V15. Upstream Kafka grants that
allowance once per request: readFromLocalLog starts minOneMessage true, clears it
as soon as any partition returns bytes, and clamps every later partition with
limitBytes, so a partition reached once the budget is spent returns zero bytes.

That zero-byte response is the only input to Kafka's fetch fairness. Both readers
of this function demote a partition only when it returned bytes --
FetchSession.PartitionIterator on recordsSize > 0, AbstractFetcherThread on
validBytes > 0 -- so a partition that returns nothing keeps its place and is served
first, with a full budget, next round. Fairness is a rotation across responses, not
a division within one.

The per-partition floor defeated that: every partition looked served, so the
rotation demoted them all and their relative order never changed. Partitions past
the budget sat at one batch per round while the head drained at full rate, turning
concurrent catch-up into serialized catch-up -- total drain time became the sum of
per-partition drain times rather than the maximum. Replaying each reader's
move-to-end rule against the real function: four lagging partitions over four
rounds were served [32, 4, 4, 4] batches where fair is [8, 8, 8, 8]; at the
consolidation fetcher's default budget ratio, ten partitions over ten rounds got
[50 x5, 10 x5] against a fair 25 each -- and spent 300 batches of work to do it
versus 250 fair, because each starved partition still paid a scan iteration and an
object GET per round to receive a batch that could not close its gap. The floor
also uncapped the response, overshooting by one batch per partition rather than one
per request: 6000 bytes returned against a 2500-byte budget at four partitions.

V27 changes one predicate, l_partition_batch_count = 0 to l_global_bytes = 0. The
bounded O(result) scan and its index-scan plan dependency, error handling, request
ordering and max_batches_per_partition are untouched -- the function body is
otherwise identical to V19. Partitions past the budget now return an empty batch
list, which FetchCompleter maps to MemoryRecords.EMPTY with Errors.NONE and the real
high watermark: the response a caught-up partition gets today, so this reaches an
existing path for a second reason.

That path had a latent gap this change surfaces. The empty branch dropped
lastStableOffset, passing OptionalLong.empty() where the served branch carries the
high watermark; that reaches the consumer as INVALID_LAST_STABLE_OFFSET and skips
SubscriptionState.updateLastStableOffset. It was harmless while only a caught-up
partition reached the branch, but under request-level budgeting a lagging partition
reaches it on every round it waits its turn, staling a READ_COMMITTED consumer's
currentLag and endOffsets. FetchCompleter now carries the high watermark as the LSO
on the empty success path, and the error path keeps it unset. Covered by
FetchCompleterTest.testEmptyResponseForLaggingPartitionCarriesLastStableOffset,
which asserts a below-watermark fetch of an empty response returns the LSO.

InMemoryControlPlane, the alternative control plane selectable for local and
system-test deployments, previously applied fetchMaxBytes per partition and ignored
both maxPartitionFetchBytes and maxBatchesPerPartition. It now enforces all three,
so AbstractControlPlaneTest can assert one expectation against both planes. Tests
that relied on the in-memory plane ignoring maxPartitionFetchBytes see fewer
batches.

This cannot be reordered before the fetch-session ordering commit it sits on.
Without that, request order is a HashMap iteration order identical on every fetch,
so the same partitions would be served every round and the rest would get zero
instead of one -- strictly worse than the floor.

Five parameterised rows across three existing tests asserted the floor; they are
relocated into findServesOnlyTheFirstPartitionWhenTheRequestBudgetIsSpent with the
second partition now expected empty. No existing expectation was adjusted in place.

The consolidation purgatory comment called the overshoot "a heap-sizing concern
only", which missed that the same rule disabled its rotation.
@jeqo
test(inkless:consolidation): pin the empty-response contract the fetc…
d1cb2f3 
…h rotation needs [KC-407]

find_batches now returns no batches for a partition once the request budget is
spent, and AbstractFetcherThread.processFetchRequest only calls
partitionStates.updateAndMoveToEnd when validBytes > 0 -- so an empty response is
what leaves an unserved consolidating partition at the front of partitionStates to
be served first next iteration. Nothing covered the path between those two facts.

The hops are all pre-existing behaviour, which is why this is test-only:
clampRecordsToSegment returns an empty block untouched, maybeStampDisklessLeaderEpoch
iterates nothing, and UnifiedLog.append early-returns on validBytes <= 0 without
rolling a segment. But clampRecordsToSegment documents "at least one batch is always
emitted for progress" -- true when trimming a non-empty block, and a plausible thing
for someone to extend into a floor for an empty one. That change would compile, pass
every existing test, and silently reinstate the starvation on the consolidation path.

Asserts a fetch whose partition the control plane served nothing comes back with
Errors.NONE, zero record bytes, and the real high watermark, so lag stays visible
while the partition waits its turn. Verified non-vacuous by mutation: making the
endpoint emit a synthetic batch for an empty block fails the record-bytes assertion.

The probe in DelayedConsolidationFetch.tryComplete runs unbudgeted --
findDisklessBatches passes maxBytes = Int.MaxValue -- so it legitimately reports data
while the budgeted fetch serves this partition nothing. That is the case modelled
here, and the reason the request-level grant does not make consolidation fetches park
on minBytes.
@jeqo
jeqo force-pushed 8000 the jeqo/fetch-fairness branch from 0b90462 to d1cb2f3 Compare August 21, 2026 10:38
AnatolyPopov
AnatolyPopov reviewed Aug 21, 2026
View reviewed changes
Comment thread core/src/main/scala/kafka/server/ReplicaManager.scala Outdated
case Some(handler) => handler.handle(params, fetchInfos.toMap.asJava).thenApply(_.asScala.toSeq)
case Some(handler) =>
val ordered = mutable.LinkedHashMap.from(fetchInfos).asJava
fetchInfos.foreach { case (tp, data) => ordered.put(tp, data) }
@AnatolyPopov AnatolyPopov Aug 21, 2026
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This I guess is not needed anymore, since the new LinkedHashMap is created with the elements of fetchInfos already

@jeqo jeqo Aug 21, 2026
Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

good catch. was too lazy to verify before pushing 😅

@AnatolyPopov
AnatolyPopov commented Aug 21, 2026
Copy link
Copy Markdown
Contributor

On top of one leftover I posted above, this comment https://github.com/aiven/inkless/blob/main/core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala#L193C1-L198C87 seems to be now outdated. Could you check it @jeqo?
Otherwise LGTM.

AnatolyPopov
AnatolyPopov approved these changes Aug 21, 2026
View reviewed changes
@AnatolyPopov AnatolyPopov left a comment
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM, thanks for addressing everything!

@AnatolyPopov
AnatolyPopov merged commit b1fcb5f into main Aug 21, 2026
6 checks passed
@AnatolyPopov
AnatolyPopov deleted the jeqo/fetch-fairness branch August 21, 2026 14:04
jeqo added a commit that referenced this pull request Sep 9, 2026
@jeqo
fix(inkless:control_plane): make the diskless fetch budget fair acros…
2e91877 
…s partitions [KC-407] (#750)

* fix(inkless): preserve fetch-session order into the diskless control plane [KC-407]

find_batches_v2 spends a fetch's shared fetch.max.bytes budget in request-array
order, and that array is built by iterating the map handed to the diskless fetch
handler. The broker had already computed a fair order: FetchSession moves every
partition that returned bytes to the back of session.partitionMap, so a starved
partition drifts to the front and is served first on the next fetch. That order
reached fetchDisklessMessages intact, as an ordered Seq.

fetchInfos.toMap discarded it. scala.collection.immutable.Map has dedicated
Map1..Map4 classes that preserve insertion order, but at five or more entries it
is a HashMap whose iteration order ignores insertion and is a pure function of
the key set -- so it is identical on every subsequent fetch for a stable
partition assignment. Hash order thereby replaced the rotation with a fixed
priority ranking keyed on partition identity.

This does not by itself make the allocation fair. The rotation it restores stays
inert while find_batches guarantees every partition at least one batch, because
FetchSession only demotes partitions that returned bytes and under that floor
they all do. This is a prerequisite for that fix, not a substitute for it.

The new test uses eight partitions deliberately; four or fewer pass even against
the discarded order.

* fix(inkless:control_plane): make find_batches minOneMessage request-level [KC-407]

find_batches_v2 admitted the first batch of every partition unconditionally --
"always >=1 batch per partition", inherited from V15. Upstream Kafka grants that
allowance once per request: readFromLocalLog starts minOneMessage true, clears it
as soon as any partition returns bytes, and clamps every later partition with
limitBytes, so a partition reached once the budget is spent returns zero bytes.

That zero-byte response is the only input to Kafka's fetch fairness. Both readers
of this function demote a partition only when it returned bytes --
FetchSession.PartitionIterator on recordsSize > 0, AbstractFetcherThread on
validBytes > 0 -- so a partition that returns nothing keeps its place and is served
first, with a full budget, next round. Fairness is a rotation across responses, not
a division within one.

The per-partition floor defeated that: every partition looked served, so the
rotation demoted them all and their relative order never changed. Partitions past
the budget sat at one batch per round while the head drained at full rate, turning
concurrent catch-up into serialized catch-up -- total drain time became the sum of
per-partition drain times rather than the maximum. Replaying each reader's
move-to-end rule against the real function: four lagging partitions over four
rounds were served [32, 4, 4, 4] batches where fair is [8, 8, 8, 8]; at the
consolidation fetcher's default budget ratio, ten partitions over ten rounds got
[50 x5, 10 x5] against a fair 25 each -- and spent 300 batches of work to do it
versus 250 fair, because each starved partition still paid a scan iteration and an
object GET per round to receive a batch that could not close its gap. The floor
also uncapped the response, overshooting by one batch per partition rather than one
per request: 6000 bytes returned against a 2500-byte budget at four partitions.

V27 changes one predicate, l_partition_batch_count = 0 to l_global_bytes = 0. The
bounded O(result) scan and its index-scan plan dependency, error handling, request
ordering and max_batches_per_partition are untouched -- the function body is
otherwise identical to V19. Partitions past the budget now return an empty batch
list, which FetchCompleter maps to MemoryRecords.EMPTY with Errors.NONE and the real
high watermark: the response a caught-up partition gets today, so this reaches an
existing path for a second reason.

That path had a latent gap this change surfaces. The empty branch dropped
lastStableOffset, passing OptionalLong.empty() where the served branch carries the
high watermark; that reaches the consumer as INVALID_LAST_STABLE_OFFSET and skips
SubscriptionState.updateLastStableOffset. It was harmless while only a caught-up
partition reached the branch, but under request-level budgeting a lagging partition
reaches it on every round it waits its turn, staling a READ_COMMITTED consumer's
currentLag and endOffsets. FetchCompleter now carries the high watermark as the LSO
on the empty success path, and the error path keeps it unset. Covered by
FetchCompleterTest.testEmptyResponseForLaggingPartitionCarriesLastStableOffset,
which asserts a below-watermark fetch of an empty response returns the LSO.

InMemoryControlPlane, the alternative control plane selectable for local and
system-test deployments, previously applied fetchMaxBytes per partition and ignored
both maxPartitionFetchBytes and maxBatchesPerPartition. It now enforces all three,
so AbstractControlPlaneTest can assert one expectation against both planes. Tests
that relied on the in-memory plane ignoring maxPartitionFetchBytes see fewer
batches.

This cannot be reordered before the fetch-session ordering commit it sits on.
Without that, request order is a HashMap iteration order identical on every fetch,
so the same partitions would be served every round and the rest would get zero
instead of one -- strictly worse than the floor.

Five parameterised rows across three existing tests asserted the floor; they are
relocated into findServesOnlyTheFirstPartitionWhenTheRequestBudgetIsSpent with the
second partition now expected empty. No existing expectation was adjusted in place.

The consolidation purgatory comment called the overshoot "a heap-sizing concern
only", which missed that the same rule disabled its rotation.

* test(inkless:consolidation): pin the empty-response contract the fetch rotation needs [KC-407]

find_batches now returns no batches for a partition once the request budget is
spent, and AbstractFetcherThread.processFetchRequest only calls
partitionStates.updateAndMoveToEnd when validBytes > 0 -- so an empty response is
what leaves an unserved consolidating partition at the front of partitionStates to
be served first next iteration. Nothing covered the path between those two facts.

The hops are all pre-existing behaviour, which is why this is test-only:
clampRecordsToSegment returns an empty block untouched, maybeStampDisklessLeaderEpoch
iterates nothing, and UnifiedLog.append early-returns on validBytes <= 0 without
rolling a segment. But clampRecordsToSegment documents "at least one batch is always
emitted for progress" -- true when trimming a non-empty block, and a plausible thing
for someone to extend into a floor for an empty one. That change would compile, pass
every existing test, and silently reinstate the starvation on the consolidation path.

Asserts a fetch whose partition the control plane served nothing comes back with
Errors.NONE, zero record bytes, and the real high watermark, so lag stays visible
while the partition waits its turn. Verified non-vacuous by mutation: making the
endpoint emit a synthetic batch for an empty block fails the record-bytes assertion.

The probe in DelayedConsolidationFetch.tryComplete runs unbudgeted --
findDisklessBatches passes maxBytes = Int.MaxValue -- so it legitimately reports data
while the budgeted fetch serves this partition nothing. That is the case modelled
here, and the reason the request-level grant does not make consolidation fetches park
on minBytes.

* fixup! fix(inkless): preserve fetch-session order into the diskless control plane [KC-407]

* fixup! fix(inkless:control_plane): make find_batches minOneMessage request-level [KC-407]
jeqo added a commit that referenced this pull request Sep 9, 2026
@jeqo
sync(test): adapt the #750 and #774 tests to the 4.1 fetch APIs
06c54f9 
DelayedFetch on 4.1 takes Seq[(TopicIdPartition, FetchPartitionStatus)] for
both partition-status parameters, not util.LinkedHashMap, so the
fetch-order test builds the diskless statuses as a Seq.

LogReadResult on 4.1 lives in org.apache.kafka.server and its constructors
end with Optional[Throwable] instead of Errors. Successful reads drop the
trailing Errors.NONE and the out-of-range read passes
OffsetOutOfRangeException, which LogReadResult.error maps back to
OFFSET_OUT_OF_RANGE.
jeqo added a commit that referenced this pull request Sep 9, 2026
@jeqo
fix(inkless:control_plane): make the diskless fetch budget fair acros…
0320342 
…s partitions [KC-407] (#750)

* fix(inkless): preserve fetch-session order into the diskless control plane [KC-407]

find_batches_v2 spends a fetch's shared fetch.max.bytes budget in request-array
order, and that array is built by iterating the map handed to the diskless fetch
handler. The broker had already computed a fair order: FetchSession moves every
partition that returned bytes to the back of session.partitionMap, so a starved
partition drifts to the front and is served first on the next fetch. That order
reached fetchDisklessMessages intact, as an ordered Seq.

fetchInfos.toMap discarded it. scala.collection.immutable.Map has dedicated
Map1..Map4 classes that preserve insertion order, but at five or more entries it
is a HashMap whose iteration order ignores insertion and is a pure function of
the key set -- so it is identical on every subsequent fetch for a stable
partition assignment. Hash order thereby replaced the rotation with a fixed
priority ranking keyed on partition identity.

This does not by itself make the allocation fair. The rotation it restores stays
inert while find_batches guarantees every partition at least one batch, because
FetchSession only demotes partitions that returned bytes and under that floor
they all do. This is a prerequisite for that fix, not a substitute for it.

The new test uses eight partitions deliberately; four or fewer pass even against
the discarded order.

* fix(inkless:control_plane): make find_batches minOneMessage request-level [KC-407]

find_batches_v2 admitted the first batch of every partition unconditionally --
"always >=1 batch per partition", inherited from V15. Upstream Kafka grants that
allowance once per request: readFromLocalLog starts minOneMessage true, clears it
as soon as any partition returns bytes, and clamps every later partition with
limitBytes, so a partition reached once the budget is spent returns zero bytes.

That zero-byte response is the only input to Kafka's fetch fairness. Both readers
of this function demote a partition only when it returned bytes --
FetchSession.PartitionIterator on recordsSize > 0, AbstractFetcherThread on
validBytes > 0 -- so a partition that returns nothing keeps its place and is served
first, with a full budget, next round. Fairness is a rotation across responses, not
a division within one.

The per-partition floor defeated that: every partition looked served, so the
rotation demoted them all and their relative order never changed. Partitions past
the budget sat at one batch per round while the head drained at full rate, turning
concurrent catch-up into serialized catch-up -- total drain time became the sum of
per-partition drain times rather than the maximum. Replaying each reader's
move-to-end rule against the real function: four lagging partitions over four
rounds were served [32, 4, 4, 4] batches where fair is [8, 8, 8, 8]; at the
consolidation fetcher's default budget ratio, ten partitions over ten rounds got
[50 x5, 10 x5] against a fair 25 each -- and spent 300 batches of work to do it
versus 250 fair, because each starved partition still paid a scan iteration and an
object GET per round to receive a batch that could not close its gap. The floor
also uncapped the response, overshooting by one batch per partition rather than one
per request: 6000 bytes returned against a 2500-byte budget at four partitions.

V27 changes one predicate, l_partition_batch_count = 0 to l_global_bytes = 0. The
bounded O(result) scan and its index-scan plan dependency, error handling, request
ordering and max_batches_per_partition are untouched -- the function body is
otherwise identical to V19. Partitions past the budget now return an empty batch
list, which FetchCompleter maps to MemoryRecords.EMPTY with Errors.NONE and the real
high watermark: the response a caught-up partition gets today, so this reaches an
existing path for a second reason.

That path had a latent gap this change surfaces. The empty branch dropped
lastStableOffset, passing OptionalLong.empty() where the served branch carries the
high watermark; that reaches the consumer as INVALID_LAST_STABLE_OFFSET and skips
SubscriptionState.updateLastStableOffset. It was harmless while only a caught-up
partition reached the branch, but under request-level budgeting a lagging partition
reaches it on every round it waits its turn, staling a READ_COMMITTED consumer's
currentLag and endOffsets. FetchCompleter now carries the high watermark as the LSO
on the empty success path, and the error path keeps it unset. Covered by
FetchCompleterTest.testEmptyResponseForLaggingPartitionCarriesLastStableOffset,
which asserts a below-watermark fetch of an empty response returns the LSO.

InMemoryControlPlane, the alternative control plane selectable for local and
system-test deployments, previously applied fetchMaxBytes per partition and ignored
both maxPartitionFetchBytes and maxBatchesPerPartition. It now enforces all three,
so AbstractControlPlaneTest can assert one expectation against both planes. Tests
that relied on the in-memory plane ignoring maxPartitionFetchBytes see fewer
batches.

This cannot be reordered before the fetch-session ordering commit it sits on.
Without that, request order is a HashMap iteration order identical on every fetch,
so the same partitions would be served every round and the rest would get zero
instead of one -- strictly worse than the floor.

Five parameterised rows across three existing tests asserted the floor; they are
relocated into findServesOnlyTheFirstPartitionWhenTheRequestBudgetIsSpent with the
second partition now expected empty. No existing expectation was adjusted in place.

The consolidation purgatory comment called the overshoot "a heap-sizing concern
only", which missed that the same rule disabled its rotation.

* test(inkless:consolidation): pin the empty-response contract the fetch rotation needs [KC-407]

find_batches now returns no batches for a partition once the request budget is
spent, and AbstractFetcherThread.processFetchRequest only calls
partitionStates.updateAndMoveToEnd when validBytes > 0 -- so an empty response is
what leaves an unserved consolidating partition at the front of partitionStates to
be served first next iteration. Nothing covered the path between those two facts.

The hops are all pre-existing behaviour, which is why this is test-only:
clampRecordsToSegment returns an empty block untouched, maybeStampDisklessLeaderEpoch
iterates nothing, and UnifiedLog.append early-returns on validBytes <= 0 without
rolling a segment. But clampRecordsToSegment documents "at least one batch is always
emitted for progress" -- true when trimming a non-empty block, and a plausible thing
for someone to extend into a floor for an empty one. That change would compile, pass
every existing test, and silently reinstate the starvation on the consolidation path.

Asserts a fetch whose partition the control plane served nothing comes back with
Errors.NONE, zero record bytes, and the real high watermark, so lag stays visible
while the partition waits its turn. Verified non-vacuous by mutation: making the
endpoint emit a synthetic batch for an empty block fails the record-bytes assertion.

The probe in DelayedConsolidationFetch.tryComplete runs unbudgeted --
findDisklessBatches passes maxBytes = Int.MaxValue -- so it legitimately reports data
while the budgeted fetch serves this partition nothing. That is the case modelled
here, and the reason the request-level grant does not make consolidation fetches park
on minBytes.

* fixup! fix(inkless): preserve fetch-session order into the diskless control plane [KC-407]

* fixup! fix(inkless:control_plane): make find_batches minOneMessage request-level [KC-407]
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

Copilot code review Copilot
Copilot review effort
Applies to this pull request for everyone.Learn more about Copilot code review.
Copilot left review comments
@AnatolyPopov AnatolyPopov AnatolyPopov approved these changes

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@jeqo @AnatolyPopov
0