Skip to content

Commit f82bb06

Browse files
viktorsomogyijeqo
viktorsomogyi
and
jeqo
authored
feat(inkless): KC-72 Reconcile stale records after diskless switch (#612)
* feat(inkless): KC-72 Reconcile stale records after diskless switch When a partition switches to diskless, a replica can still hold uncommitted records above the high watermark. If the switch is quick and consolidation is enabled, the consolidation fetchers would start appending on top of that stale tail. This commit implements the following: - Truncate a newly switched partition's local log down to the just-committed seal (classic-to-diskless start offset). This runs after makeLeader/makeFollower and before any fetcher starts, so catch-up and consolidation always initialize from the truncated LEO. - Add ConsolidationReconciler to decide when a consolidating partition joins the consolidation fetcher, covering both the initial switch and resume after failover, restart, or reassignment. ReplicaManager now delegates consolidation fetcher startup to it. - Hand off from the classic ReplicaFetcher to consolidation once a follower reaches the seal, via the fetcher's self-eviction path. - Gate pruning of switched diskless partitions behind a prune floor and skip partitions whose switch is still pending; born-consolidated partitions keep pruning directly. - Transform the per-partition diskless watermark into a single monotonic safeConsolidationPruningFloor (gate + progress tracker), replacing lastAppliedDisklessLogStartOffset. * Apply suggestions from code review Co-authored-by: Jorge Esteban Quilcate Otoya <jorge.quilcate@aiven.io> * Address review comments * Fix failing tests Some tests were incorrectly using a LEO that is < seal offset and therefore the most recent review change where we marked such partitions fenced, failed. * Update core/src/main/scala/kafka/server/ReplicaManager.scala Co-authored-by: Jorge Esteban Quilcate Otoya <jorge.quilcate@aiven.io> --------- Co-authored-by: Jorge Esteban Quilcate Otoya <jorge.quilcate@aiven.io>
1 parent 0136f3d commit f82bb06

12 files changed

Lines changed: 983 additions & 102 deletions

‎core/src/main/scala/io/aiven/inkless/consolidation/ConsolidatedDisklessLogPruner.scala‎

Lines changed: 38 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -19,46 +19,46 @@
1919
package io.aiven.inkless.consolidation
2020

2121
import io.aiven.inkless.control_plane.{ControlPlane, PruneDisklessLogsError, PruneDisklessLogsRequest}
22+
import kafka.cluster.Partition
2223
import kafka.server.ReplicaManager
2324
import kafka.server.metadata.InklessMetadataView
2425
import kafka.utils.Logging
2526
import org.apache.kafka.common.TopicIdPartition
27+
import org.apache.kafka.metadata.PartitionRegistration
2628

2729
import scala.jdk.CollectionConverters.{CollectionHasAsScala, SeqHasAsJava}
2830

2931
class ConsolidatedDisklessLogPruner(replicaManager: ReplicaManager,
3032
inklessMetadataView: InklessMetadataView,
3133
controlPlane: ControlPlane) extends Runnable with Logging {
3234

33-
/**
34-
* This method is repeatedly invoked until the thread shuts down or this method throws an exception
35-
*/
3635
override def run(): Unit = {
37-
val disklessTopicIdPartitions = inklessMetadataView.getConsolidatingDisklessTopicPartitions.asScala
38-
val eitherErrorOrLog = disklessTopicIdPartitions
39-
.map(tip => replicaManager.getPartitionOrError(tip.topicPartition))
40-
.partition(either => either.isLeft)
41-
eitherErrorOrLog._1
42-
.flatMap {
43-
case Left(error) => Some(error)
44-
case _ => None
45-
}
36+
// Read the classic-to-diskless start offset once per partition and thread it through, so the
37+
// eligibility check and the per-partition prune decision always see the same value (no TOCTOU
38+
// between dropping SWITCH_PENDING and computing the safe prune offset).
39+
val eligiblePartitionsWithSeal = inklessMetadataView.getConsolidatingDisklessTopicPartitions.asScala
40+
.map(tip => (tip, inklessMetadataView.getClassicToDisklessStartOffset(tip.topicPartition)))
41+
.filter { case (_, seal) => seal != PartitionRegistration.CLASSIC_TO_DISKLESS_SWITCH_PENDING }
42+
.map { case (tip, seal) => (replicaManager.getPartitionOrError(tip.topicPartition), seal) }
43+
eligiblePartitionsWithSeal
44+
.collect { case (Left(error), _) => error }
4645
.foreach(error => logger.warn("Got error during pruning consolidated diskless logs: {}", error.message))
47-
val requests = eitherErrorOrLog._2
48-
.flatMap {
49-
case Right(partition) =>
50-
partition.topicId.flatMap { topicId =>
51-
partition.log.flatMap { log =>
52-
val highestRemoteOffset = log.highestOffsetInRemoteStorage
53-
if (highestRemoteOffset < 0) {
54-
None
55-
} else {
46+
val requests = eligiblePartitionsWithSeal
47+
.collect { case (Right(partition), seal) => (partition, seal) }
48+
.flatMap { case (partition, seal) =>
49+
partition.topicId.flatMap { topicId =>
50+
partition.log.flatMap { log =>
51+
val highestRemoteOffset = log.highestOffsetInRemoteStorage
52+
if (highestRemoteOffset < 0) {
53+
None
54+
} else {
55+
safePruneOffset(partition, seal, highestRemoteOffset).map { safeHighestRemoteOffset =>
5656
val topicIdPartition = new TopicIdPartition(topicId, partition.topicPartition)
57-
Some(new PruneDisklessLogsRequest(topicIdPartition, highestRemoteOffset))
57+
new PruneDisklessLogsRequest(topicIdPartition, safeHighestRemoteOffset)
5858
}
5959
}
6060
}
61-
case _ => None
61+
}
6262
}.toSeq.asJava
6363
if (!requests.isEmpty) {
6464
controlPlane.pruneDisklessLogs(requests).asScala.foreach { pruneDisklessLogsResponse =>
@@ -70,7 +70,7 @@ class ConsolidatedDisklessLogPruner(replicaManager: ReplicaManager,
7070
replicaManager.getPartitionOrError(pruneDisklessLogsResponse.topicIdPartition.topicPartition) match {
7171
case Right(partition) =>
7272
val newDisklessLogStart = pruneDisklessLogsResponse.disklessLogStartOffset
73-
partition.maybeAdvanceLastAppliedDisklessLogStartOffset(newDisklessLogStart)
73+
partition.maybeAdvanceConsolidationPruneFloor(newDisklessLogStart)
7474
case Left(error) => logger.warn("Couldn't update diskless start offset for {} due to: {}",
7575
pruneDisklessLogsResponse.topicIdPartition.topicPartition,
7676
error.message
@@ -80,4 +80,18 @@ class ConsolidatedDisklessLogPruner(replicaManager: ReplicaManager,
8080
}
8181
}
8282
}
83+
84+
private def safePruneOffset(partition: Partition, seal: Long, highestRemoteOffset: Long): Option[Long] = {
85+
seal match {
86+
case PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET =>
87+
Some(highestRemoteOffset)
88+
case classicToDisklessStartOffset if classicToDisklessStartOffset >= 0 =>
89+
partition.getSafeConsolidatedDisklessPruneOffset(highestRemoteOffset)
90+
case unexpected =>
91+
logger.warn("Skipping pruning for {} due to unexpected classic-to-diskless start offset {}",
92+
partition.topicPartition,
93+
unexpected)
94+
None
95+
}
96+
}
8397
}

‎core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationFetcherThread.scala‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,8 @@ class ConsolidationFetcherThread(name: String,
4343
}
4444
}
4545

46+
override protected def shouldEvictFullySwitchedDisklessPartitions: Boolean = false
47+
4648
override def processPartitionData(
4749
topicPartition: TopicPartition,
4850
fetchOffset: Long,

‎core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationReconciler.scala‎

Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
/*
2+
* Inkless
3+
* Copyright (C) 2024 - 2026 Aiven OY
4+
*
5+
* This program is free software: you can redistribute it and/or modify
6+
* it under the terms of the GNU Affero General Public License as published by
7+
* the Free Software Foundation, either version 3 of the License, or
8+
* (at your option) any later version.
9+
*
10+
* This program is distributed in the hope that it will be useful,
11+
* but WITHOUT ANY WARRANTY; without even the implied warranty of
12+
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13+
* GNU Affero General Public License for more details.
14+
*
15+
* You should have received a copy of the GNU Affero General Public License
16+
* along with this program. If not, see <http://www.gnu.org/licenses/>.
17+
*/
18+
19+
package io.aiven.inkless.consolidation
20+
21+
import kafka.cluster.Partition
22+
import kafka.server.{InitialFetchState, ReplicaManager}
23+
import kafka.server.metadata.InklessMetadataView
24+
import org.apache.kafka.common.TopicPartition
25+
import org.apache.kafka.logger.StateChangeLogger
26+
import org.apache.kafka.metadata.PartitionRegistration
27+
import org.apache.kafka.server.network.BrokerEndPoint
28+
import org.apache.kafka.storage.internals.log.UnifiedLog
29+
30+
import scala.collection.{Set, mutable}
31+
import scala.jdk.OptionConverters.RichOptional
32+
33+
/**
34+
* Outcome of per-partition reconciliation before consolidation can start.
35+
* - Ready: the partition is safe to hand to the consolidation fetcher at the given offset.
36+
* - Retry: the partition cannot consolidate yet (e.g., pending seal, LEO below seal) — skip
37+
* this round; a later metadata delta or classic-fetcher catch-up will re-trigger.
38+
* - Failed: an unrecoverable error; mark the partition as failed in the fetcher manager.
39+
*/
40+
private sealed trait ConsolidationStartState
41+
private object ConsolidationStartState {
42+
final case class Ready(offset: Long) extends ConsolidationStartState
43+
final case class Retry(reason: String) extends ConsolidationStartState
44+
final case class Failed(reason: Throwable) extends ConsolidationStartState
45+
}
46+
47+
private class ReconciliationException(message: String) extends RuntimeException(message)
48+
49+
class ConsolidationReconciler(replicaManager: ReplicaManager,
50+
stateChangeLogger: StateChangeLogger,
51+
consolidationMetrics: ConsolidationMetrics,
52+
inklessMetadataView: InklessMetadataView,
53+
initialFetchOffset: UnifiedLog => Long,
54+
consolidationFetcherManager: ConsolidationFetcherManager) {
55+
56+
/**
57+
* Starts the consolidation fetchers for the given partitions in the parameter. These partitions
58+
* must be ready for consolidation in order to be started (meaning LEO == seal offset).
59+
* If a partition wasn't ready for consolidation because of some error or the LEO for the given
60+
* partition was behind the seal offset, then it will be logged and won't be started.
61+
*
62+
* @param consolidatingPartitions the consolidating partitions to start fetching.
63+
*/
64+
def startConsolidationFetchers(consolidatingPartitions: mutable.HashMap[TopicPartition, Partition]): Unit = {
65+
if (consolidatingPartitions.nonEmpty) {
66+
val consolidatingPartitionAndOffsets: mutable.HashMap[TopicPartition, InitialFetchState] =
67+
initConsolidatingPartitionFetching(consolidatingPartitions)
68+
69+
consolidationFetcherManager.addFetcherForPartitions(consolidatingPartitionAndOffsets)
70+
consolidatingPartitionAndOffsets.keys.foreach(tp => consolidationMetrics.registerPartition(tp))
71+
}
72+
}
73+
74+
def startConsolidationFetchersForCaughtUpClassicPartitions(topicPartitions: Set[TopicPartition]): Unit = {
75+
val consolidatingDisklessPartitionsToStartFetching = new mutable.HashMap[TopicPartition, Partition]
76+
topicPartitions.foreach { tp =>
77+
if (inklessMetadataView.isConsolidatingDisklessTopic(tp.topic)) {
78+
replicaManager.onlinePartition(tp).foreach(partition => consolidatingDisklessPartitionsToStartFetching.put(tp, partition))
79+
}
80+
}
81+
startConsolidationFetchers(consolidatingDisklessPartitionsToStartFetching)
82+
}
83+
84+
def initConsolidatingPartitionFetching(consolidatingDisklessPartitionsToStartFetching: mutable.HashMap[TopicPartition, Partition]
85+
): mutable.HashMap[TopicPartition, InitialFetchState] = {
86+
val consolidatingPartitionAndOffsets = new mutable.HashMap[TopicPartition, InitialFetchState]
87+
88+
consolidatingDisklessPartitionsToStartFetching.foreachEntry { (topicPartition, partition) =>
89+
val log = partition.localLogOrException
90+
reconcileSwitchedConsolidatingDisklessPartition(partition) match {
91+
case ConsolidationStartState.Ready(offset) =>
92+
consolidatingPartitionAndOffsets.put(topicPartition, InitialFetchState(
93+
log.topicId.toScala,
94+
new BrokerEndPoint(-1, "diskless", -1),
95+
partition.getLeaderEpoch,
96+
offset
97+
))
98+
case ConsolidationStartState.Retry(reason) =>
99+
stateChangeLogger.info(reason)
100+
case ConsolidationStartState.Failed(reason) =>
101+
stateChangeLogger.error("Error happened during consolidating log reconciliation before initial fetch from diskless control plane", reason)
102+
consolidationFetcherManager.addFailedPartition(topicPartition)
103+
}
104+
}
105+
consolidatingPartitionAndOffsets
106+
}
107+
108+
private def reconcileSwitchedConsolidatingDisklessPartition(partition: Partition): ConsolidationStartState = {
109+
val tp = partition.topicPartition
110+
inklessMetadataView.getClassicToDisklessStartOffset(tp) match {
111+
case PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET =>
112+
// Born-diskless/born-consolidated topics don't have a classic seal boundary to reconcile.
113+
ConsolidationStartState.Ready(initialFetchOffset(partition.localLogOrException))
114+
case PartitionRegistration.CLASSIC_TO_DISKLESS_SWITCH_PENDING =>
115+
ConsolidationStartState.Retry(s"Skipping consolidation for $tp because classic-to-diskless migration is still pending")
116+
case seal if seal >= 0 =>
117+
val log = partition.localLogOrException
118+
if (log.logEndOffset < seal) {
119+
// The classic prefix hasn't been fully replicated locally yet; a classic catch-up
120+
// fetcher must finish bringing the local log up to the seal before consolidation
121+
// can take over.
122+
ConsolidationStartState.Retry(
123+
s"Skipping consolidation for $tp because local LEO ${log.logEndOffset} is below " +
124+
s"classic-to-diskless start offset $seal")
125+
} else {
126+
// LEO >= seal. This covers both the initial switch (LEO == seal, nothing consolidated
127+
// yet) and resuming an already-progressed partition after a restart, leadership
128+
// failover, or reassignment (the local log either kept its consolidated frontier or
129+
// was rehydrated from tiered storage). In every case we resume from the current local
130+
// LEO so we never re-consolidate or skip data the local log already holds.
131+
//
132+
// The prune floor is the higher of the seal and the current log start offset:
133+
// - at first switch logStartOffset is still the classic prefix start, so the floor is
134+
// the seal, which blocks pruning the diskless region until consolidation has tiered
135+
// past the boundary;
136+
// - on resume logStartOffset has advanced past the seal as consolidated segments were
137+
// tiered and deleted, so it reflects real pruning progress.
138+
val pruneFloor = math.max(seal, log.logStartOffset)
139+
partition.ensureConsolidationPruneFloorAtLeast(pruneFloor)
140+
ConsolidationStartState.Ready(log.logEndOffset)
141+
}
142+
case unexpected =>
143+
ConsolidationStartState.Failed(new ReconciliationException(s"Skipping consolidation for $tp due to unexpected classic-to-diskless start offset: $unexpected"))
144+
}
145+
}
146+
}

‎core/src/main/scala/io/aiven/inkless/consolidation/DisklessLeaderEndPoint.scala‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -179,7 +179,6 @@ class DisklessLeaderEndPoint(
179179
val job = fetchOffsetHandler.createJob()
180180
val futures = mutable.Map.empty[TopicPartition, CompletableFuture[FileRecordsOrError]]
181181

182-
// TODO: POD-2419, offsets for leader epoch needs to be implemented for proper truncation
183182
partitions.forEach { (tp, epochData) =>
184183
if (epochData.leaderEpoch != OffsetsForLeaderEpochResponse.UNDEFINED_EPOCH && job.mustHandle(tp.topic)) {
185184
val partitionRequest = new ListOffsetsPartition()

‎core/src/main/scala/kafka/cluster/Partition.scala‎

Lines changed: 45 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -219,10 +219,14 @@ class Partition(val topicPartition: TopicPartition,
219219
}
220220
}
221221

222-
// Broker-local watermark for the latest diskless log start offset applied from pruning responses.
223-
// The control plane owns the persisted log start offset; this value only catches stale or
224-
// out-of-order pruner responses before they can regress local partition state.
225-
@volatile private var lastAppliedDisklessLogStartOffset: Long = PartitionRegistration.NO_CLASSIC_TO_DISKLESS_START_OFFSET
222+
// Monotonic prune watermark for a consolidating diskless partition. It plays two roles:
223+
// - gate: the consolidated diskless region must not be pruned until tiered (remote) storage has
224+
// caught up to this offset, so pruning is only allowed once highestRemoteOffset >= floor;
225+
// - progress tracker: it advances to the latest applied diskless log start offset as pruning
226+
// proceeds, guarding against stale or out-of-order pruner responses regressing local state.
227+
// None means the partition has no consolidation prune boundary established yet (so it must not be
228+
// pruned on the switched-partition path).
229+
@volatile private var safeConsolidationPruningFloor: Option[Long] = None
226230

227231
this.logIdent = s"[Partition $topicPartition broker=$localBrokerId] "
228232

@@ -264,18 +268,48 @@ class Partition(val topicPartition: TopicPartition,
264268
*/
265269
def isAtMinIsr: Boolean = leaderLogIfLocal.exists { partitionState.isr.size == effectiveMinIsr(_) }
266270

267-
def maybeAdvanceLastAppliedDisklessLogStartOffset(newDisklessLogStartOffset: Long): Unit = {
271+
/**
272+
* Establishes a lower bound on the consolidation prune floor when consolidation (re)starts.
273+
* Keeping an already-higher floor (e.g. the pruner advanced it while logStartOffset still points
274+
* at the classic prefix) is the expected steady state, so this is silent on a no-op.
275+
*/
276+
def ensureConsolidationPruneFloorAtLeast(floor: Long): Unit = {
268277
inWriteLock(leaderIsrUpdateLock) {
269-
if (newDisklessLogStartOffset >= lastAppliedDisklessLogStartOffset) {
270-
lastAppliedDisklessLogStartOffset = newDisklessLogStartOffset
271-
} else {
272-
warn(s"Ignoring stale diskless log start offset for $topicPartition. " +
273-
s"The new value ($newDisklessLogStartOffset) is less than the last locally applied value " +
274-
s"($lastAppliedDisklessLogStartOffset).")
278+
raiseConsolidationPruneFloor(floor)
279+
}
280+
}
281+
282+
/**
283+
* Advances the consolidation prune floor as pruning progresses. A value below the current floor
284+
* indicates a stale or out-of-order pruner response, which is unexpected and surfaced as a warning.
285+
*/
286+
def maybeAdvanceConsolidationPruneFloor(newFloor: Long): Unit = {
287+
inWriteLock(leaderIsrUpdateLock) {
288+
if (!raiseConsolidationPruneFloor(newFloor)) {
289+
warn(s"Ignoring stale consolidation prune floor for $topicPartition. " +
290+
s"The new value ($newFloor) is less than the current floor " +
291+
s"(${safeConsolidationPruningFloor.get}).")
275292
}
276293
}
277294
}
278295

296+
// Raises the floor to `floor` if it does not regress an existing one. Returns whether the floor
297+
// was applied. Callers must hold leaderIsrUpdateLock.
298+
private def raiseConsolidationPruneFloor(floor: Long): Boolean = {
299+
if (safeConsolidationPruningFloor.forall(floor >= _)) {
300+
safeConsolidationPruningFloor = Some(floor)
301+
true
302+
} else {
303+
false
304+
}
305+
}
306+
307+
def getSafeConsolidatedDisklessPruneOffset(highestRemoteOffset: Long): Option[Long] = {
308+
safeConsolidationPruningFloor
309+
.filter(floor => highestRemoteOffset >= floor)
310+
.map(_ => highestRemoteOffset)
311+
}
312+
279313
def isSealed: Boolean = _sealed
280314

281315
/**

‎core/src/main/scala/kafka/server/ReplicaFetcherThread.scala‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,15 @@ class ReplicaFetcherThread(name: String,
102102
evictFullySwitchedDisklessPartitions()
103103
}
104104

105+
/**
106+
* Whether the eviction check in processPartitionData should mark fully-switched partitions
107+
* for removal. The classic ReplicaFetcherThread enables this (so it self-evicts at the seal
108+
* and hands off to consolidation). The ConsolidationFetcherThread disables it because it
109+
* intentionally fetches for already-switched partitions.
110+
*/
111+
112+
protected def shouldEvictFullySwitchedDisklessPartitions: Boolean = true
113+
105114
// process fetched data
106115
override def processPartitionData(
107116
topicPartition: TopicPartition,
@@ -157,7 +166,9 @@ class ReplicaFetcherThread(name: String,
157166
// has committed a classicToDisklessStartOffset for this partition AND our local LEO has reached it,
158167
// the follower is fully caught up to the leader's frozen classic log and must not keep fetching.
159168
val classicToDisklessStartOffset = replicaMgr.inklessMetadataView().getClassicToDisklessStartOffset(topicPartition)
160-
if (classicToDisklessStartOffset >= 0 && log.logEndOffset >= classicToDisklessStartOffset) {
169+
if (shouldEvictFullySwitchedDisklessPartitions &&
170+
classicToDisklessStartOffset >= 0 &&
171+
log.logEndOffset >= classicToDisklessStartOffset) {
161172
partitionsToEvictAfterDisklessSwitch += topicPartition
162173
}
163174

@@ -178,6 +189,7 @@ class ReplicaFetcherThread(name: String,
178189
info(s"Evicting partitions from this replica fetcher because they have completed the " +
179190
s"classic-to-diskless switch and the local log has caught up to the seal offset: $toEvict")
180191
replicaMgr.replicaFetcherManager.removeFetcherForPartitions(toEvict)
192+
replicaMgr.startConsolidationFetchersForCaughtUpClassicPartitions(toEvict)
181193
}
182194
}
183195

0 commit comments

Comments
 (0)