Problem

At the local/diskless boundary of a consolidating partition, consumers fetching with large minBytes would park in the delayed-fetch purgatory and wait the full maxWaitMs before getting a response — even when diskless data was already available past the local logEndOffset.

Solution

When the local log read doesn't satisfy minBytes, supplement with a synchronous diskless fetch starting at logEndOffset and merge the results via ConcatenatedRecords.concat().

For requests spanning only consolidating partitions, the supplement is applied inline and the response is produced immediately without entering the purgatory. For mixed requests (consolidating + pure-diskless), the supplement and the diskless fetch run concurrently in DelayedFetch.onComplete so latency equals the slower of the two, not their sum.

Changes

Inline supplement in fetchMessages

• Fires when consolidating partitions are below minBytes and no pure-diskless partitions are present
• Guards: !isFromFollower, maxWaitMs > 0, !hasPreferredReadReplica, !errorReadingData
• inklessSharedState checked at collection time so supplements are never gathered on a non-inkless broker
• bytesReadable only counts error-free supplement bytes — a supplement response with an error cannot trick the broker into responding prematurely
• Exception handling: InterruptedException re-interrupts the thread; TimeoutException logged specifically for operator visibility; other failures fall back to local-only

Concurrent supplement in DelayedFetch.onComplete

• Supplement and pure-diskless fetches run in parallel via CompletableFuture.thenCombine
• AtomicBoolean ensures the response callback fires exactly once
• Skips partitions with local read errors or fetchOffset >= logEndOffset
• Skips entirely for follower fetches
• On diskless future failure, per-partition error entries (Errors.forException) are returned so clients see the partition in the response rather than it being silently dropped

ConcatenatedRecords.concat(prefix, tail) static factory

• Encapsulates type dispatch over MemoryRecords/ConcatenatedRecords
• Unwraps inner backing records directly — no extra materialization copy
• Validates inputs with Objects.requireNonNull

buildConsolidationSupplementFetchInfos

• Computes per-partition remaining byte budget (original maxBytes minus local bytes read)
• Drops partitions with zero budget remaining

mergeConsolidationSupplement

• Materializes local FileRecords → MemoryRecords (standard readInto/flip idiom)
• Promotes diskless HW/LSO; uses Math.min(local, supplement) for logStartOffset
• Preserves localData.abortedTransactions (warns when present since diskless has no transaction index)
• Preserves localData.preferredReadReplica (diskless returns empty for this field)
• Falls back to local-only data on unknown Records types or merge failures with error log

Tests

• testFetchConsolidatingSupplementRespondsWithoutDelayWhenDisklessDataAvailable: RED/GREEN — fails without the supplement (parks for maxWaitMs), passes with it (immediate response, purgatory stays empty)
• testFetchConsolidatingParksUntilMaxWaitWhenNoDisklessSupplementData: bounds the contract — no diskless data means the request correctly waits
• testFetchConsolidatingSupplementWithErrorDoesNotInflateBytesReadable: supplement with error status must not satisfy minBytes — request still parks
• testFetchConsolidatingSupplementPreservesInterruptedStatus: thread interrupted status is preserved after InterruptedException in supplement fetch
• testFetchMixedConsolidatingPureDisklessAndClassicPartitions: all three partition types in one request, asserts merged consolidating data and two diskless handler calls
• testCompletionWithConsolidationSupplement: focused unit test for the concurrent supplement path in DelayedFetch.onComplete
• testConsolidationSupplementSkippedWhenLocalReadHasError: supplement merge is not applied when local read returns an error
• testConsolidationSupplementNotIssuedWhenLocalReadHasError: supplement fetch is not issued for error-result partitions
• testDisklessFetchFailureReturnsPerPartitionErrors: diskless future failure produces per-partition error entries, not silent omission
• Unit tests for buildConsolidationSupplementFetchInfos: budget deduction, zero-budget drop, missing partitions filtered
• Unit tests for mergeConsolidationSupplement: MemoryRecords merge, FileRecords materialization, unknown-type fallback