fix(inkless:consolidation): defer prefix reads on a rebuilding replica [KC-673] by jeqo · Pull Request #847 · aiven/inkless

jeqo · 2026-11-02T11:08:48Z

A consolidating replica that has not rebuilt its tier state answers consumer fetches of the remote-only prefix with OFFSET_OUT_OF_RANGE, and the consumer resets its position. With auto.offset.reset=latest it skips every record up to the log end. This happens in two places: the diskless fetch of a replica with an empty local log, and the remote read of a replica that rebuilds again after the WAL start moves on (the rebuild clears the leader epoch cache before it restores it).

Both places now answer a consumer fetch of a retained offset with the retriable OFFSET_NOT_AVAILABLE. Replica fetches and deleted offsets keep OFFSET_OUT_OF_RANGE. The deferral has no time limit, like a classic follower that retries a failed rebuild. A ReplicaManager meter replaces the under-replicated signal. A second metric counts tier rebuild requests, so a replica that the WAL prune keeps outrunning is visible. This PR doesn't change the rebuild or the prune. A rebuild loop is found with that metric and stopped with existing settings, which the operator guide lists.

Commits

test: integration repro that wipes a follower and reads the prefix for none and latest.
fix: the mapping on both paths, the LeaderEpochNotAvailableException thrown by RemoteLogManager.read, the deferral meter, ConsolidationTierRebuildRequests, the operator guide, and unit tests.

See the commit messages for the reasoning behind each step.

Metrics and operator impact

New meter under kafka.server:type=ReplicaManager: ConsolidationDeferredPrefixFetchRate, the rate of consumer fetches answered with OFFSET_NOT_AVAILABLE. A rate over zero for longer than a rebuild takes points at a replica that does not recover. It drops to zero when consumers stop reading.
New per-partition counter and broker aggregate under io.aiven.inkless.consolidation:type=ConsolidationMetrics: ConsolidationTierRebuildRequests.
One INFO line for the first deferral after a quiet minute. The consolidation fetcher already logs why a rebuild does not finish.
DISKLESS_CONSOLIDATION.md documents the deferral and the settings that stop a rebuild loop.

Testing

InklessConsolidatedDisklessTopicsTest.testWipedReplicaServesRemoteOnlyPrefixOfConsolidatedTopic wipes a follower, restarts it, and reads from offset 0 with a consumer pinned to it. It fails on the diskless path without the fix (none and latest).
ReplicaManagerInklessTest pins the mapping on each path: the range bounds, replica fetches, other errors, non-consolidating topics, and the deferral meter.
DisklessLeaderEndPointTest and ConsolidationFetcherThreadTest cover the rebuild counter.

Validation

Covered: unit tests for the mapping on both paths, and a two-broker integration test that wipes a replica and reads the prefix.
Not covered: a multi-node or system test, and a run with the WAL prune and rebuilds active at the same time.
Known limits: the deferral has no time limit, and the bound for a deleted offset on the diskless path can lag (the cross-tier earliest cache), so a deleted offset below a stale bound defers until the bound catches up. The remote path defers only a read that finds no leader epoch for the offset (new LeaderEpochNotAvailableException, a subclass of OffsetOutOfRangeException). A read with a known epoch and no remote segment keeps OFFSET_OUT_OF_RANGE, as on a classic tiered topic. The docs list these limits.
Rollout check: watch ConsolidationDeferredPrefixFetchRate and ConsolidationTierRebuildRequests while TSU is enabled. A rate that stays above zero for longer than a rebuild takes needs a look.

Follow-ups

ListOffsets by timestamp on a rebuilding replica: separate PR.
Keep consumers off a replica that has not rebuilt (routing readiness).
Change the rebuild or the prune (hold the WAL prune for a slow replica, or resume the rebuild from the newest uploaded segment). Not planned. Revisit only if ConsolidationTierRebuildRequests shows a loop that the settings don't stop.

Copilot

🟡 Changes recommended

Deferred-partition retries trigger quadratic expiration scans in recovery callbacks, and the regression test can pass without exercising a fetch against an unrebuilt replica.

2 open issues

What changed in this PR

This PR prevents consumers from resetting past retained remote-prefix records while a consolidating replica rebuilds its tier state.

Changes:

Return retriable OFFSET_NOT_AVAILABLE on diskless and remote prefix-read paths.
Add deferral gauges, a tier-rebuild request count, and operator guidance.
Add unit coverage and a wiped-replica integration test for consumer reset policies.

File	Description
`docs/inkless/DISKLESS_CONSOLIDATION.md`	Document prefix deferral, metrics, and rebuild-loop mitigation.
`core/src/test/scala/unit/kafka/server/ReplicaManagerInklessTest.scala`	Test error mapping, range boundaries, and deferral gauge lifecycle.
`core/src/test/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPointTest.scala`	Verify rebuild request counting.
`core/src/test/scala/io/aiven/inkless/consolidation/ConsolidationFetcherThreadTest.scala`	Test rebuild metric aggregation and registration lifecycle.
`core/src/test/java/kafka/server/InklessConsolidatedDisklessTopicsTest.java`	Exercise remote-prefix reads during wiped-replica recovery.
`core/src/main/scala/kafka/server/ReplicaManager.scala`	Defer retained-prefix reads and track deferral episodes.
`core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala`	Record tier-rebuild requests.
`core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationMetrics.scala`	Expose partition and broker rebuild counts.
`checkstyle/suppressions.xml`	Add complexity suppressions for the integration test class.

_{Review effort: Balanced}

Copilot

🟡 Changes recommended

The restart wait is skipped if polling, position lookup, or an assertion throws, allowing teardown to close the controller and delete files while startup is still running; move the wait into a finally block.

2 open issues

1 resolved since last review

Avoid scanning all episodes for each deferred partition

_{Review effort: Balanced}

Copilot

🟡 Changes recommended

The advertised deferred-prefix gauges are not registered, so alerts using those names receive no samples; the implementation or alert guidance needs to be aligned.

2 open issues

1 resolved since last review

Wait for restart completion even when polling fails

_{Review effort: Balanced}

Copilot

🔵 Needs a closer look

Unlimited recovery retries interact with concurrent WAL pruning and tier-state rebuilds, requiring running-cluster validation of retained-offset recovery and deleted-offset handling before approval.

1 open issue

Advertised consolidation deferral gauges are not registered

1 resolved since last review

Ensure the test observes a fetch before rebuild completes

_{Review effort: Balanced}

Copilot

🔵 Needs a closer look

Consumer-offset safety remains unverified when asynchronous tier rebuilds race with WAL pruning, requiring running-cluster validation of retained-offset recovery and deleted-offset handling.

No open issues 🎉

1 resolved since last review

Advertised consolidation deferral gauges are not registered

_{Review effort: Balanced}

…a [KC-673] Once a consolidating diskless partition is tiered and its WAL pruned, the offsets below the WAL start live only in remote storage. A replica with an empty local log, such as a reassignment target or a replica that lost its data, routes fetches below the WAL start to the diskless path until its consolidation fetcher rebuilds the tier state. The control plane answers OFFSET_OUT_OF_RANGE there, and metadata routing can send a consumer to that replica during the window. The test wipes a follower's partition data, restarts it, and reads from offset 0 with a consumer pinned to it, once per auto.offset.reset policy. It asserts that the read never fails with OFFSET_OUT_OF_RANGE and never moves the position past 0. On this commit none and latest fail; the fix follows in a later commit. Assisted-by: Claude Opus 5.5

…a [KC-673] A consolidating replica that rebuilds its tier state answers consumer fetches below the WAL start with OFFSET_OUT_OF_RANGE from two places. With auto.offset.reset=latest the consumer then skips every record up to the log end. The diskless path serves a replica with an empty local log. fetchDisklessMessages maps the answer to the retriable OFFSET_NOT_AVAILABLE for a consumer fetch when the offset is below the WAL start and at or above the cross-tier earliest offset. An unknown earliest offset defers too. A supplement fetch passes deferPrefixReads = false, because the merge drops its errors. The remote path serves a replica that rebuilds again. The rebuild clears the leader epoch cache before it restores it, and a read in between finds no epoch. RemoteLogManager.read now throws the subclass LeaderEpochNotAvailableException for that case only. A known epoch with no remote segment keeps OffsetOutOfRangeException, as on a classic tiered topic. processRemoteFetch maps the subclass to OFFSET_NOT_AVAILABLE when the offset is at or above the local log start. It makes no control-plane call on the reader thread. Replica fetches and deleted offsets keep OFFSET_OUT_OF_RANGE. The deferral has no time limit, and the diskless ISR has no under-replicated signal for it. The ReplicaManager meter ConsolidationDeferredPrefixFetchRate counts deferred fetches on both paths. The broker logs one INFO line after a quiet minute. The WAL prune doesn't read the fetch position of any replica, so a slow replica can rebuild again after each prune. The new per-partition counter ConsolidationTierRebuildRequests counts each OFFSET_MOVED_TO_TIERED_STORAGE answer on the broker of the replica. A failed rebuild that retries counts again. The operator guide lists the settings that stop a loop. It changes no rebuild or prune code. Assisted-by: Claude Opus 5.5

giuseppelillo · 2026-11-04T09:28:53Z

+                                    disklessStartOffset: Long): Boolean =
+    // Doesn't read the local log: the rebuild can finish between routing and this answer.
+    isFromConsumer && fetchOffset < disklessStartOffset && consolidationActiveFor(tp.topicPartition()) && {
+      val earliest = crossTierEarliestOffset(tp.topicPartition())


On a cache miss, crossTierEarliestOffset makes a synchronous controlPlane().listOffsets call. Here it runs inside thenApply, so it lands on the Reader fetch thread, once per matching partition. The remote path avoids this on purpose. Could this read only the cache? A miss already counts as unknown and defers.

Good catch. shouldDeferPrefixRead now reads only the cache, and a miss counts as unknown and defers. I extended the cache-miss test to check that listOffsets is never called.

giuseppelillo · 2026-11-04T09:28:53Z

                    logger.debug("Offset {} for {} is below the diskless WAL start {} but at/above the whole-log start {}; " +
                      "signalling OFFSET_MOVED_TO_TIERED_STORAGE to rebuild the consolidated remote prefix.",
                      requestedOffset, tp.topicPartition, disklessStartOffset, logStartOffset)
+                    replicaManager.recordConsolidationTierRebuildRequest(tp.topicPartition)


The consolidation fetcher uses TierStateMachine, which already marks BuildRemoteLogAuxStateRequestsPerSec and BuildRemoteLogAuxStateErrorsPerSec per topic. The new counter only adds per-partition detail. We could at least mention the upstream meters in the guide: the errors meter separates a failing rebuild from a prune loop without the log grep.

Agreed. Added both upstream meters to the rebuild-loop section of the guide.

giuseppelillo · 2026-11-04T09:28:53Z

+  private def recordPrefixDeferral(tp: TopicIdPartition, fetchOffset: Long, disklessStartOffset: Option[Long]): Unit = {
+    consolidationDeferredPrefixFetchRate.mark()
+    val now = time.milliseconds
+    val lastMs = lastPrefixDeferralMs.getAndSet(now)


Nit: lastPrefixDeferralMs is one broker-wide timestamp, and every deferral updates it. A partition that keeps deferring silences this line for every other partition, and a replica that stays stuck logs once. If the timestamp updates only when the line logs, the broker logs at most once a minute instead.

Agreed. The timestamp now updates only when the line logs, so the broker logs at most once a minute.

giuseppelillo · 2026-11-04T09:28:54Z

+   * True for a consumer fetch on a consolidating topic with
+   * `crossTierEarliest <= fetchOffset < disklessStartOffset`; false otherwise.
+   * An unknown `crossTierEarliest` defers too: it is unknown only until the control plane answers.
+   * Replica fetches keep the error, because the consolidation fetcher starts the rebuild from it.


Nit: the consolidation fetcher never reaches fetchDisklessMessages. It calls fetchHandler.handle through DelayedConsolidationFetch. The consumer-only guard is right, but the reason is that only a consumer resets on this error. The comment at line 9770 in ReplicaManagerInklessTest has the same wording.

You're right. The fetcher uses DelayedConsolidationFetch, not this path. Both comments reworded

giuseppelillo · 2026-11-04T09:28:54Z

+ * example during a tier state rebuild. A read whose epoch is known but has no remote segment throws
+ * a plain {@link OffsetOutOfRangeException}.
+ */
+public class LeaderEpochNotAvailableException extends OffsetOutOfRangeException {


Nit: this new class sits in an upstream package, and the PR also edits RemoteLogManager.java. Neither file is in INKLESS_OWNERSHIP. Add the class as owned and RemoteLogManager.java as interleaved.

Good catch. Added both entries to INKLESS_OWNERSHIP.

… replica [KC-673] Assisted-by: Claude Sonnet 5.5

…a [KC-673] (#847) * test(inkless:consolidation): reproduce prefix reads on a wiped replica [KC-673] Once a consolidating diskless partition is tiered and its WAL pruned, the offsets below the WAL start live only in remote storage. A replica with an empty local log, such as a reassignment target or a replica that lost its data, routes fetches below the WAL start to the diskless path until its consolidation fetcher rebuilds the tier state. The control plane answers OFFSET_OUT_OF_RANGE there, and metadata routing can send a consumer to that replica during the window. The test wipes a follower's partition data, restarts it, and reads from offset 0 with a consumer pinned to it, once per auto.offset.reset policy. It asserts that the read never fails with OFFSET_OUT_OF_RANGE and never moves the position past 0. On this commit none and latest fail; the fix follows in a later commit. Assisted-by: Claude Opus 5.5 * fix(inkless:consolidation): defer prefix reads on a rebuilding replica [KC-673] A consolidating replica that rebuilds its tier state answers consumer fetches below the WAL start with OFFSET_OUT_OF_RANGE from two places. With auto.offset.reset=latest the consumer then skips every record up to the log end. The diskless path serves a replica with an empty local log. fetchDisklessMessages maps the answer to the retriable OFFSET_NOT_AVAILABLE for a consumer fetch when the offset is below the WAL start and at or above the cross-tier earliest offset. An unknown earliest offset defers too. A supplement fetch passes deferPrefixReads = false, because the merge drops its errors. The remote path serves a replica that rebuilds again. The rebuild clears the leader epoch cache before it restores it, and a read in between finds no epoch. RemoteLogManager.read now throws the subclass LeaderEpochNotAvailableException for that case only. A known epoch with no remote segment keeps OffsetOutOfRangeException, as on a classic tiered topic. processRemoteFetch maps the subclass to OFFSET_NOT_AVAILABLE when the offset is at or above the local log start. It makes no control-plane call on the reader thread. Replica fetches and deleted offsets keep OFFSET_OUT_OF_RANGE. The deferral has no time limit, and the diskless ISR has no under-replicated signal for it. The ReplicaManager meter ConsolidationDeferredPrefixFetchRate counts deferred fetches on both paths. The broker logs one INFO line after a quiet minute. The WAL prune doesn't read the fetch position of any replica, so a slow replica can rebuild again after each prune. The new per-partition counter ConsolidationTierRebuildRequests counts each OFFSET_MOVED_TO_TIERED_STORAGE answer on the broker of the replica. A failed rebuild that retries counts again. The operator guide lists the settings that stop a loop. It changes no rebuild or prune code. Assisted-by: Claude Opus 5.5

On 4.1, `RemoteStorageFetchInfo` carries a `TopicPartition` instead of a `TopicIdPartition`, so the remote fetch callback in `ReplicaManager` looks up the topic ID with `topicIdPartition`. `LogReadResult` takes no `Errors` argument, `RetriableRemoteStorageException` doesn't exist and the reclaim floor throws `RemoteStorageException`, and `remote.log.manager.thread.pool.size` sizes the follower pool. `HostedPartition.Online` and `HostedPartition.Offline` take no type parameter, `MemoryRecords` and `SimpleRecord` are in `org.apache.kafka.common.record`, and the Scala `BrokerReconfigurable` returns a Scala set. The changes added by #831, #834, #835, #840, #841, #847, #850, and #852 used the main signatures. The Java `UnifiedLogTest` on 4.1 is a stub, and the log tests still live in the Scala `kafka.log.UnifiedLogTest`. The cherry-pick of #840 ports its four `forceDeleteLocalSegmentsBelow` tests to the Scala suite instead of pulling the upstream Java tests in through the conflict. After #847, `kafka.server.ReplicaManager` on 4.1 is too big for SpotBugs to analyze, and `:core:spotbugsMain` fails with `SKIPPED_CLASS_TOO_BIG`. The class is about 400 lines longer on 4.1 than on 4.2 and 4.3, and SpotBugs 4.9.8 fails the same way. The exclude filter now skips that pattern for `ReplicaManager` only. Assisted-by: Claude Opus 5.5

jeqo requested a balanced review from Copilot November 2, 2026 11:10

Copilot started reviewing on behalf of jeqo November 2, 2026 11:11

View session