feat(inkless): add lag metrics for consolidated topic partitions

core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationFetcherManager.scala

@@ -27,7 +27,8 @@ class ConsolidationFetcherManager(brokerConfig: KafkaConfig,
                                   replicaManager: ReplicaManager,
                                   quotaManager: ReplicationQuotaManager,
                                   fetchHandler: FetchHandler,
-                                  fetchOffsetHandler: FetchOffsetHandler)
+                                  fetchOffsetHandler: FetchOffsetHandler,
+                                  consolidationMetrics: Option[ConsolidationMetrics] = None)
   extends AbstractFetcherManager[ConsolidationFetcherThread](name = "ConsolidationFetcherManager on broker " + brokerConfig.brokerId,
     clientId = "Consolidation",
     numFetchers = brokerConfig.numReplicaFetchers) {
@@ -47,7 +48,7 @@ class ConsolidationFetcherManager(brokerConfig: KafkaConfig,
       replicaManager.brokerEpochSupplier
     )
     new ConsolidationFetcherThread(threadName, disklessLeaderEndPoint, brokerConfig, failedPartitions, replicaManager,
-      quotaManager, logContext.logPrefix)
+      quotaManager, logContext.logPrefix, consolidationMetrics)
   }
 
   def shutdown(): Unit = {

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

@@ -20,21 +20,52 @@ package io.aiven.inkless.consolidation
 
 import io.aiven.inkless.consume.ConcatenatedRecords
 import kafka.server.{FailedPartitions, KafkaConfig, ReplicaFetcherThread, ReplicaManager, ReplicaQuota}
+import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.record.{MemoryRecords, Records}
 import org.apache.kafka.server.LeaderEndPoint
+import org.apache.kafka.storage.internals.log.LogAppendInfo
+
+import scala.util.Try
 
 class ConsolidationFetcherThread(name: String,
                                  leader: LeaderEndPoint,
                                  brokerConfig: KafkaConfig,
                                  failedPartitions: FailedPartitions,
                                  replicaMgr: ReplicaManager,
                                  quota: ReplicaQuota,
-                                 logPrefix: String) extends ReplicaFetcherThread(name, leader, brokerConfig, failedPartitions, replicaMgr, quota, logPrefix) {
+                                 logPrefix: String,
+                                 consolidationMetrics: Option[ConsolidationMetrics] = None) extends ReplicaFetcherThread(name, leader, brokerConfig, failedPartitions, replicaMgr, quota, logPrefix) {
 
   override def toMemoryRecords(records: Records): MemoryRecords = {
     (records: @unchecked) match {
       case r: ConcatenatedRecords => r.toMemoryRecords
       case _ => super.toMemoryRecords(records)
     }
   }
+
+  override def processPartitionData(
+    topicPartition: TopicPartition,
+    fetchOffset: Long,
+    partitionLeaderEpoch: Int,
+    partitionData: FetchData
+  ): Option[LogAppendInfo] = {
+    val result = super.processPartitionData(topicPartition, fetchOffset, partitionLeaderEpoch, partitionData)
+
+    consolidationMetrics.foreach { metrics =>
+      val logOpt = Try(replicaMgr.getPartitionOrException(topicPartition).localLogOrException).toOption
+      logOpt.foreach { log =>
+        val disklessLogEndOffset = partitionData.highWatermark
+        val localLogEndOffset = log.logEndOffset
+        val remoteLogEndOffset = log.highestOffsetInRemoteStorage()
+
+        metrics.updateLocalLag(topicPartition, Math.max(0L, disklessLogEndOffset - localLogEndOffset))
+        if (remoteLogEndOffset >= 0) {
+          metrics.updateLag(topicPartition, Math.max(0L, disklessLogEndOffset - remoteLogEndOffset))
+          metrics.updateDeletableMessages(topicPartition, Math.max(0L, remoteLogEndOffset - log.localLogStartOffset()))
+        }
+      }
+    }
+
+    result
+  }
 }

core/src/main/scala/io/aiven/inkless/consolidation/ConsolidationMetrics.scala

@@ -0,0 +1,82 @@
+/*
+ * Inkless
+ * Copyright (C) 2024 - 2026 Aiven OY
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package io.aiven.inkless.consolidation
+
+import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.server.metrics.KafkaMetricsGroup
+
+import java.io.Closeable
+import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicLong
+import scala.jdk.CollectionConverters._
+
+class ConsolidationMetrics extends Closeable {
+  private val Lag = "inkless.remote.consolidation.lag"
+  private val LocalLag = "inkless.remote.consolidation.local.lag"
+  private val DeletableMessages = "inkless.remote.consolidation.deletable.messages"
+
+  private val metricsGroup = new KafkaMetricsGroup("io.aiven.inkless.consolidation", "ConsolidationMetrics")
+
+  private val lagByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
+  private val localLagByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
+  private val deletableByPartition = new ConcurrentHashMap[TopicPartition, AtomicLong]()
+
+  def registerPartition(tp: TopicPartition): Unit = {
+    val tags = Map("topic" -> tp.topic, "partition" -> tp.partition.toString).asJava
+
+    lagByPartition.computeIfAbsent(tp, _ => {
+      val value = new AtomicLong(0)
+      metricsGroup.newGauge(Lag, () => value.get, tags)
+      value
+    }).set(0)
+    localLagByPartition.computeIfAbsent(tp, _ => {
+      val value = new AtomicLong(0)
+      metricsGroup.newGauge(LocalLag, () => value.get, tags)
+      value
+    }).set(0)
+    deletableByPartition.computeIfAbsent(tp, _ => {
+      val value = new AtomicLong(0)
+      metricsGroup.newGauge(DeletableMessages, () => value.get, tags)
+      value
+    }).set(0)
+  }
+
+  def updateLag(tp: TopicPartition, lag: Long): Unit =
+    Option(lagByPartition.get(tp)).foreach(_.set(lag))
+
+  def updateLocalLag(tp: TopicPartition, lag: Long): Unit =
+    Option(localLagByPartition.get(tp)).foreach(_.set(lag))
+
+  def updateDeletableMessages(tp: TopicPartition, count: Long): Unit =
+    Option(deletableByPartition.get(tp)).foreach(_.set(count))
+
+  def unregisterPartition(tp: TopicPartition): Unit = {
+    val tags = Map("topic" -> tp.topic, "partition" -> tp.partition.toString).asJava
+    lagByPartition.remove(tp)
+    localLagByPartition.remove(tp)
+    deletableByPartition.remove(tp)
+    metricsGroup.removeMetric(Lag, tags)
+    metricsGroup.removeMetric(LocalLag, tags)
+    metricsGroup.removeMetric(DeletableMessages, tags)
+  }
+
+  override def close(): Unit = {
+    lagByPartition.keys.asScala.toList.foreach(unregisterPartition)
+  }
+}

core/src/main/scala/kafka/server/ReplicaManager.scala

@@ -23,7 +23,7 @@ import io.aiven.inkless.control_plane.{BatchInfo, FindBatchRequest, FindBatchRes
 import io.aiven.inkless.delete.{DeleteRecordsInterceptor, FileCleaner, RetentionEnforcer}
 import io.aiven.inkless.merge.FileMerger
 import io.aiven.inkless.produce.AppendHandler
-import io.aiven.inkless.consolidation.ConsolidationFetcherManager
+import io.aiven.inkless.consolidation.{ConsolidationFetcherManager, ConsolidationMetrics}
 import kafka.cluster.Partition
 import kafka.log.LogManager
 import kafka.server.HostedPartition.Online
@@ -284,13 +284,18 @@ class ReplicaManager(val config: KafkaConfig,
   private val inklessFileCleaner: Option[FileCleaner] = inklessSharedState.map(new FileCleaner(_))
   // FIXME: FileMerger is having issues with hanging queries. Disabling until fixed.
   private val inklessFileMerger: Option[FileMerger] = None // inklessSharedState.map(new FileMerger(_))
+  private val consolidationMetrics: Option[ConsolidationMetrics] =
+    if (config.disklessRemoteStorageConsolidationEnabled && inklessFetchHandler.isDefined && inklessFetchOffsetHandler.isDefined)
+      Some(new ConsolidationMetrics())
+    else
+      None
   private val consolidationFetcherManager: Option[ConsolidationFetcherManager] =
     if (config.disklessRemoteStorageConsolidationEnabled) {
       if (inklessFetchHandler.isEmpty || inklessFetchOffsetHandler.isEmpty) {
         logger.warn("Remote storage consolidation is enabled, however Inkless doesn't seem to be configured properly.")
       }
       inklessFetchHandler.zip(inklessFetchOffsetHandler).map { case (fetchHandler, fetchOffsetHandler) =>
-        new ConsolidationFetcherManager(config, this, quotaManagers.follower, fetchHandler, fetchOffsetHandler)
+        new ConsolidationFetcherManager(config, this, quotaManagers.follower, fetchHandler, fetchOffsetHandler, consolidationMetrics)
       }
     } else {
       None
@@ -487,8 +492,13 @@ class ReplicaManager(val config: KafkaConfig,
     val partitions = partitionsToStop.map(_.topicPartition)
     replicaFetcherManager.removeFetcherForPartitions(partitions)
     replicaAlterLogDirsManager.removeFetcherForPartitions(partitions)
-    consolidationFetcherManager.foreach(_.removeFetcherForPartitions(
-      partitions.filter(p => _inklessMetadataView.isConsolidatingDisklessTopic(p.topic))))
+    val consolidatingPartitionsToStop = if (config.disklessRemoteStorageConsolidationEnabled)
+      partitions.filter(p => _inklessMetadataView.isConsolidatingDisklessTopic(p.topic))
+      else Set[TopicPartition]()
+    consolidationFetcherManager.foreach(_.removeFetcherForPartitions(consolidatingPartitionsToStop))
+    consolidationMetrics.foreach { metrics =>
+      consolidatingPartitionsToStop.foreach(tp => metrics.unregisterPartition(tp))
+    }
 
     // Second remove deleted partitions from the partition map. Fetchers rely on the
     // ReplicaManager to get Partition's information so they must be stopped first.
@@ -2590,6 +2600,9 @@ class ReplicaManager(val config: KafkaConfig,
       replicaFetcherManager.removeFetcherForPartitions(newOfflinePartitions)
       replicaAlterLogDirsManager.removeFetcherForPartitions(newOfflinePartitions ++ partitionsWithOfflineFutureReplica.map(_.topicPartition))
       consolidationFetcherManager.foreach(_.removeFetcherForPartitions(newOfflinePartitions))
+      consolidationMetrics.foreach { metrics =>
+        newOfflinePartitions.foreach(tp => metrics.unregisterPartition(tp))
+      }
 
       partitionsWithOfflineFutureReplica.foreach(partition => partition.removeFutureLocalReplica(deleteFromLogDir = false))
       newOfflinePartitions.foreach { topicPartition =>
@@ -2645,6 +2658,7 @@ class ReplicaManager(val config: KafkaConfig,
     if (checkpointHW)
       checkpointHighWatermarks()
     consolidationFetcherManager.foreach(_.shutdown())
+    consolidationMetrics.foreach(_.close())
     replicaSelectorPlugin.foreach(_.close)
     removeAllTopicMetrics()
     addPartitionsToTxnManager.foreach(_.shutdown())
@@ -3055,6 +3069,9 @@ class ReplicaManager(val config: KafkaConfig,
       }
 
       consolidationFetcherManager.foreach(_.addFetcherForPartitions(consolidatingPartitionAndOffsets))
+      consolidationMetrics.foreach { metrics =>
+        consolidatingPartitionAndOffsets.keys.foreach(tp => metrics.registerPartition(tp))
+      }
       stateChangeLogger.info(s"Started consolidating diskless fetchers as part of become-leader for ${consolidatingDisklessPartitionsToStartFetching.size} partitions")
     }
   }
@@ -3186,6 +3203,9 @@ class ReplicaManager(val config: KafkaConfig,
 
       replicaFetcherManager.addFetcherForPartitions(partitionAndOffsets)
       consolidationFetcherManager.foreach(_.addFetcherForPartitions(consolidatingPartitionAndOffsets))
+      consolidationMetrics.foreach { metrics =>
+        consolidatingPartitionAndOffsets.keys.foreach(tp => metrics.registerPartition(tp))
+      }
       stateChangeLogger.info(s"Started fetchers as part of become-follower for ${partitionsToStartFetching.size} partitions")
 
       partitionsToStartFetching.foreach{ case (topicPartition, partition) =>