fix(queue): clamp kafka lag to low watermark to prevent phantom backlog (#4129)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

services/apps/cron_service/src/jobs/queueMonitoring.job.ts

@@ -189,9 +189,9 @@ async function getTopicMessageCount(
   try {
     const topicOffsets = await admin.fetchTopicOffsets(topic)
 
-    // Sum up all partition offsets to get total messages
+    // Sum (high - low) per partition — actual messages currently in the topic.
     const totalMessages = topicOffsets.reduce((sum, partition) => {
-      return sum + Number(partition.offset)
+      return sum + Math.max(0, Number(partition.offset) - Number(partition.low))
     }, 0)
 
     return totalMessages

services/libs/queue/src/vendors/kafka/client.ts

@@ -18,6 +18,7 @@ import {
 
 import { configMap } from './config'
 import { IKafkaChannelConfig, IKafkaQueueStartOptions } from './types'
+import { getKafkaMessageCounts } from './utils'
 
 export class KafkaQueueService extends LoggerBase implements IQueue {
   private readonly MAX_RECONNECT_ATTEMPTS = 5
@@ -56,65 +57,14 @@ export class KafkaQueueService extends LoggerBase implements IQueue {
 
   async getQueueMessageCount(conf: IKafkaChannelConfig): Promise<number> {
     const topic = conf.name
-    // The consumer group ID is the same as the topic name
     const groupId = topic
 
     const admin = this.client.admin()
     await admin.connect()
 
     try {
-      this.log.debug({ topic, groupId }, 'Fetching message count for topic and consumer group')
-
-      const topicOffsets = await admin.fetchTopicOffsets(topic)
-      this.log.debug({ topic, groupId, topicOffsets }, 'Topic offsets fetched')
-
-      const offsetsResponse = await admin.fetchOffsets({
-        groupId: groupId,
-        topics: [topic],
-      })
-      this.log.debug({ topic, groupId, offsetsResponse }, 'Consumer group offsets fetched')
-
-      const offsets = offsetsResponse[0].partitions
-      this.log.debug({ topic, groupId, offsets }, 'Consumer group offsets')
-
-      let totalLeft = 0
-      for (const offset of offsets) {
-        const topicOffset = topicOffsets.find((p) => p.partition === offset.partition)
-        if (topicOffset) {
-          // If consumer offset is -1, it means no committed offset, so lag is the total messages in partition
-          if (offset.offset === '-1') {
-            const lag = Number(topicOffset.offset) - Number(topicOffset.low)
-            totalLeft += lag
-            this.log.debug(
-              {
-                partition: offset.partition,
-                topicOffset: topicOffset.offset,
-                topicLow: topicOffset.low,
-                consumerOffset: offset.offset,
-                lag,
-              },
-              'Partition lag calculated (no committed offset)',
-            )
-          } else {
-            const lag = Number(topicOffset.offset) - Number(offset.offset)
-            totalLeft += lag
-            this.log.debug(
-              {
-                partition: offset.partition,
-                topicOffset: topicOffset.offset,
-                consumerOffset: offset.offset,
-                lag,
-              },
-              'Partition lag calculated',
-            )
-          }
-        } else {
-          this.log.debug({ partition: offset.partition }, 'No topic offset found for partition')
-        }
-      }
-
-      this.log.debug({ topic, groupId, totalLeft }, 'Total messages left calculated')
-      return totalLeft
+      const counts = await getKafkaMessageCounts(this.log, admin, topic, groupId)
+      return counts.unconsumed
     } catch (err) {
       this.log.error({ topic, groupId, err }, 'Failed to get message count!')
       throw err

services/libs/queue/src/vendors/kafka/utils.ts

@@ -5,8 +5,9 @@ import { Logger } from '@crowd/logging'
 /**
  * Returns the total, consumed, and unconsumed message counts for a Kafka topic/consumer-group pair.
  *
- * Handles the `-1` offset edge case: when a consumer group has never committed an offset,
- * Kafka returns `-1`. In that case all messages from the low watermark onward are unconsumed.
+ * Uses the low watermark as a floor when computing lag so that stale committed offsets
+ * (e.g. after a topic is recreated or truncated) do not produce phantom lag.
+ * A committed offset of -1 (no commit yet) is treated as if the consumer is at the low watermark.
  */
 export async function getKafkaMessageCounts(
   log: Logger,
@@ -30,16 +31,17 @@ export async function getKafkaMessageCounts(
     for (const offset of offsets) {
       const topicOffset = topicOffsets.find((p) => p.partition === offset.partition)
       if (topicOffset) {
-        totalMessages += Number(topicOffset.offset)
+        const high = Number(topicOffset.offset)
+        const low = Number(topicOffset.low)
+        // Clamp committed to [low, high]: stale commits below the floor are treated as "at floor";
+        // commits above the high watermark (topic recreated at lower offsets) are treated as "at high".
+        // -1 means no committed offset; treat as at low watermark.
+        const committedRaw = offset.offset === '-1' ? low : Number(offset.offset)
+        const committed = Math.min(Math.max(committedRaw, low), high)
 
-        if (offset.offset === '-1') {
-          // No committed offset yet — treat all messages from the low watermark as unconsumed.
-          consumedMessages += 0
-          totalLeft += Number(topicOffset.offset) - Number(topicOffset.low)
-        } else {
-          consumedMessages += Number(offset.offset)
-          totalLeft += Number(topicOffset.offset) - Number(offset.offset)
-        }
+        totalMessages += Math.max(0, high - low)
+        consumedMessages += Math.max(0, committed - low)
+        totalLeft += Math.max(0, high - committed)
       }
     }