giuseppelillo · Sep 17, 2026 · Sep 16, 2026 · Sep 17, 2026 · Sep 17, 2026 · Sep 17, 2026
diff --git a/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationFetcherManager.scala b/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationFetcherManager.scala
@@ -20,9 +20,13 @@ package io.aiven.inkless.consolidation
 
 import io.aiven.inkless.consume.{FetchHandler, FetchOffsetHandler}
 import kafka.server.{AbstractFetcherManager, KafkaConfig, ReplicaManager, ReplicationQuotaManager}
+import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.utils.LogContext
+import org.apache.kafka.server.PartitionFetchState
 import org.apache.kafka.server.network.BrokerEndPoint
 
+import scala.collection.{Map, Set}
+
 class ConsolidationFetcherManager(brokerConfig: KafkaConfig,
                                   replicaManager: ReplicaManager,
                                   quotaManager: ReplicationQuotaManager,
@@ -51,6 +55,14 @@ class ConsolidationFetcherManager(brokerConfig: KafkaConfig,
       quotaManager, logContext.logPrefix, consolidationMetrics)
   }
 
+  // Bumps the unknown-latch generation when the consolidation fetcher drops a partition, including
+  // fence and Failed/Retry paths that do not call registerPartition again.
+  override def removeFetcherForPartitions(partitions: Set[TopicPartition]): Map[TopicPartition, PartitionFetchState] = {
+    val removed = super.removeFetcherForPartitions(partitions)
+    consolidationMetrics.foreach(m => partitions.foreach(m.bumpRemotePrefixGeneration))
+    removed
+  }
+
   def shutdown(): Unit = {
     info("shutting down")
     closeAllFetchers()

diff --git a/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationMetrics.scala b/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationMetrics.scala
@@ -37,30 +37,38 @@ import scala.jdk.CollectionConverters._
  *   Only updated when remote storage is active (highestOffsetInRemoteStorage ≥ 0); stays at 0 otherwise.
  * - ConsolidationDeletableMessages: messages already in remote storage eligible for WAL pruning
  * - ConsolidationOversizedBatch: count of consolidation append attempts rejected because the block held a
- *   batch larger than the partition's segment.bytes (RecordBatchTooLargeException). The partition is parked
+ *   batch larger than the partition's `segment.bytes` (RecordBatchTooLargeException). The partition is parked
  *   and retried with backoff, so this increments once per retry while the condition persists. Non-zero
- *   indicates a partition holding a batch larger than the current segment.bytes (e.g. max.message.bytes
- *   lowered below segment.bytes over time, or a coalesced diskless unit); raise segment.bytes above the
+ *   indicates a partition holding a batch larger than the current `segment.bytes`; raise `segment.bytes` above the
  *   batch size to resume. See DisklessLeaderEndPoint.clampRecordsToSegment.
+ * - ConsolidationRemotePrefixUnknown: 1 while a WAL-gap fetch is waiting on RLMM to decide OFFSET_MOVED
+ *   vs never-tiered OFFSET_OUT_OF_RANGE (unregistered, not ready, list failed, or a covering segment
+ *   still transitional); 0 once decided.
+ *   Broker aggregate is the count of waiting partitions. Alert semantics: the
+ *   `ConsolidationRemotePrefixUnknown` row in `docs/inkless/DISKLESS_CONSOLIDATION.md`.
  */
 class ConsolidationMetrics extends Closeable {
   private val TotalLag = "ConsolidationTotalLag"
   private val LocalLag = "ConsolidationLocalLag"
   private val DeletableMessages = "ConsolidationDeletableMessages"
   private val OversizedBatch = "ConsolidationOversizedBatch"
+  private val RemotePrefixUnknown = "ConsolidationRemotePrefixUnknown"
 
   private val metricsGroup = new KafkaMetricsGroup("io.aiven.inkless.consolidation", "ConsolidationMetrics")
 
   private val totalLagByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
   private val localLagByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
   private val deletableByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
   private val oversizedBatchByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
+  private val remotePrefixUnknownByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
+  private val remotePrefixGenerationByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
 
   // Broker-level aggregate gauges (sum across all partitions)
   metricsGroup.newGauge(TotalLag, () => sumValues(totalLagByPartition))
   metricsGroup.newGauge(LocalLag, () => sumValues(localLagByPartition))
   metricsGroup.newGauge(DeletableMessages, () => sumValues(deletableByPartition))
   metricsGroup.newGauge(OversizedBatch, () => sumValues(oversizedBatchByPartition))
+  metricsGroup.newGauge(RemotePrefixUnknown, () => sumValues(remotePrefixUnknownByPartition))
 
   def registerPartition(tp: TopicPartition): Unit = {
     val tags = Map("topic" -> tp.topic, "partition" -> tp.partition.toString).asJava
@@ -86,6 +94,14 @@ class ConsolidationMetrics extends Closeable {
       metricsGroup.newGauge(OversizedBatch, () => value.get, tags)
       value
     })
+    // Resets to 0 on re-arm so a previous WAL-gap wait does not survive re-registration.
+    // Bumps the generation so an in-flight fetch from the previous arm cannot restore the latch.
+    remotePrefixGenerationByPartition.computeIfAbsent(tp, _ => new AtomicLong(0L)).incrementAndGet()
+    remotePrefixUnknownByPartition.computeIfAbsent(tp, _ => {
+      val value = new AtomicLong(0)
+      metricsGroup.newGauge(RemotePrefixUnknown, () => value.get, tags)
+      value
+    }).set(0)
   }
 
   def updateTotalLag(tp: TopicPartition, lag: Long): Unit =
@@ -100,26 +116,73 @@ class ConsolidationMetrics extends Closeable {
   def recordOversizedBatch(tp: TopicPartition): Unit =
     Option(oversizedBatchByPartition.get(tp)).foreach(_.incrementAndGet())
 
+  def remotePrefixGeneration(tp: TopicPartition): Long =
+    Option(remotePrefixGenerationByPartition.get(tp)).map(_.get).getOrElse(0L)
+
+  // Called when the consolidation fetcher drops a partition. Classic partitions never registered
+  // here, so this is a no-op for them.
+  def bumpRemotePrefixGeneration(tp: TopicPartition): Unit = {
+    Option(remotePrefixGenerationByPartition.get(tp)).foreach { generation =>
+      generation.incrementAndGet()
+      Option(remotePrefixUnknownByPartition.get(tp)).foreach(_.set(0L))
+    }
+  }
+
+  def setRemotePrefixUnknown(tp: TopicPartition, unknown: Boolean): Unit =
+    setRemotePrefixUnknown(tp, unknown, remotePrefixGeneration(tp))
+
+  def setRemotePrefixUnknown(tp: TopicPartition, unknown: Boolean, generation: Long): Unit = {
+    if (!generationMatches(tp, generation)) {
+      return
+    }
+    if (unknown) {
+      val tags = Map("topic" -> tp.topic, "partition" -> tp.partition.toString).asJava
+      // unknown=true creates the gauge if needed so a WAL-gap fetch can mark the wait.
+      remotePrefixUnknownByPartition.computeIfAbsent(tp, _ => {
+        val value = new AtomicLong(0)
+        metricsGroup.newGauge(RemotePrefixUnknown, () => value.get, tags)
+        value
+      }).set(1L)
+    } else {
+      Option(remotePrefixUnknownByPartition.get(tp)).foreach(_.set(0L))
+    }
+    if (!generationMatches(tp, generation)) {
+      Option(remotePrefixUnknownByPartition.get(tp)).foreach(_.set(0L))
+    }
+  }
+
+  private def generationMatches(tp: TopicPartition, generation: Long): Boolean =
+    Option(remotePrefixGenerationByPartition.get(tp)).exists(_.get == generation)
+
   def unregisterPartition(tp: TopicPartition): Unit = {
+    // Bump and keep the generation counter. Deleting it would let re-register reuse 1
+    // and an in-flight fetch that captured 1 could restore the latch after stopPartitions.
+    bumpRemotePrefixGeneration(tp)
     val tags = Map("topic" -> tp.topic, "partition" -> tp.partition.toString).asJava
     totalLagByPartition.remove(tp)
     localLagByPartition.remove(tp)
     deletableByPartition.remove(tp)
     oversizedBatchByPartition.remove(tp)
+    remotePrefixUnknownByPartition.remove(tp)
     metricsGroup.removeMetric(TotalLag, tags)
     metricsGroup.removeMetric(LocalLag, tags)
     metricsGroup.removeMetric(DeletableMessages, tags)
     metricsGroup.removeMetric(OversizedBatch, tags)
+    metricsGroup.removeMetric(RemotePrefixUnknown, tags)
   }
 
   override def close(): Unit = {
     // Using same keys to unregister all partition-level metrics
-    totalLagByPartition.keys.asScala.toList.foreach(unregisterPartition)
+    val partitions = (totalLagByPartition.keys.asScala ++ remotePrefixUnknownByPartition.keys.asScala ++
+      remotePrefixGenerationByPartition.keys.asScala).toSet
+    partitions.foreach(unregisterPartition)
+    remotePrefixGenerationByPartition.clear()
     // Unregistering aggregated metrics
     metricsGroup.removeMetric(TotalLag)
     metricsGroup.removeMetric(LocalLag)
     metricsGroup.removeMetric(DeletableMessages)
     metricsGroup.removeMetric(OversizedBatch)
+    metricsGroup.removeMetric(RemotePrefixUnknown)
   }
 
   private def sumValues(map: ConcurrentHashMap[TopicPartition, AtomicLong]): Long =

diff --git a/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationReconciler.scala b/core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationReconciler.scala
@@ -51,12 +51,11 @@ private class ReconciliationException(message: String) extends RuntimeException(
  *
  * ==Invariant: diskless.enable and remote.storage.enable are set together==
  * 
- * The classic-to-diskless switch enforces this invariant atomically:
- * a switched topic has both `diskless.enable=true` AND `remote.storage.enable=true` set together in the
- * same controller batch. Therefore, a diskless topic is always consolidating (tiering to remote storage).
- *
- * This reconciler assumes the invariant holds. A topic that violates it (e.g., from pre-atomic-switch metadata)
- * is marked Failed rather than silently dropping or growing the log unbounded.
+ * A consolidating topic has both `diskless.enable=true` and `remote.storage.enable=true`. The
+ * classic-to-diskless switch sets them in the same controller batch. Enabling
+ * `remote.storage.enable=true` on an already-diskless topic does the same: the controller
+ * co-commits a leader-epoch bump so this reconciler runs. A topic that violates the invariant
+ * (remote off after a switch, from pre-atomic-switch metadata) is marked Failed.
  *
  * @see LogConfigTest.scala header for the full state machine and valid transitions
  */
@@ -81,14 +80,17 @@ class ConsolidationReconciler(replicaManager: ReplicaManager,
       val consolidatingPartitionAndOffsets: mutable.HashMap[TopicPartition, InitialFetchState] =
         initConsolidatingPartitionFetching(consolidatingPartitions)
 
-      // Mark topics throttled BEFORE starting fetchers: addFetcherForPartitions starts the threads
-      // immediately and bytes only count toward the quota while the partition is already throttled,
-      // so marking after would let the first fetch bypass the quota. All consolidating topics are
-      // marked unconditionally. We never removeThrottle on stop (matching the classic ReplicaFetcher
-      // pattern); the leftover topic -> List(-1) entries are tiny and bounded, so the residue is benign.
+      // Mark topics throttled and register per-partition gauges before starting fetchers:
+      // addFetcherForPartitions starts the threads immediately, so anything the first fetch depends
+      // on has to be in place. Bytes only count toward the quota while the partition is already
+      // throttled. registerPartition resets ConsolidationRemotePrefixUnknown to 0 before the first
+      // WAL-gap fetch can set it to 1. All consolidating topics are marked throttled
+      // unconditionally. We never removeThrottle on stop (matching the classic ReplicaFetcher
+      // pattern); the leftover topic -> List(-1) entries are tiny and bounded, so the residue is
+      // benign.
       consolidatingPartitionAndOffsets.keys.map(_.topic).toSet.foreach((topic: String) => consolidationQuotaManager.markThrottled(topic))
-      consolidationFetcherManager.addFetcherForPartitions(consolidatingPartitionAndOffsets)
       consolidatingPartitionAndOffsets.keys.foreach(tp => consolidationMetrics.registerPartition(tp))
+      consolidationFetcherManager.addFetcherForPartitions(consolidatingPartitionAndOffsets)
     }
   }
 
@@ -142,9 +144,9 @@ class ConsolidationReconciler(replicaManager: ReplicaManager,
       case seal if seal >= 0 && !inklessMetadataView.isRemoteStorageEnabled(tp.topic) =>
         // Unsupported state: a switched topic (seal >= 0) with remote storage off violates the invariant.
         // This can only come from metadata written before the atomic switch enforcement.
-        // Mark Failed (not Retry) to prevent unbounded log growth — FailedPartitionsCount metric surfaces it.
-        // Recovery: operator must set remote.storage.enable=true and trigger a leader-epoch increment
-        // (restart/reassignment/preferred-leader-election) to re-run reconciliation.
+        // Mark Failed to prevent unbounded log growth. FailedPartitionsCount surfaces it.
+        // Recovery: set remote.storage.enable=true. The controller co-commits a leader-epoch bump
+        // so reconciliation runs again.
         ConsolidationStartState.Failed(new ReconciliationException(
           s"Diskless topic $tp has remote storage disabled but was switched from classic " +
             s"(violates diskless.enable implies remote.storage.enable); consolidation cannot start " +