fix: calculate kafka topic uncosummed messages correctly (#3430)

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

Author: themarolt

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

@@ -204,10 +204,19 @@ async function getMessageCounts(
       if (topicOffset) {
         // Total messages is the latest offset
         totalMessages += Number(topicOffset.offset)
-        // Consumed messages is the consumer group's offset
-        consumedMessages += Number(offset.offset)
-        // Unconsumed is the difference
-        totalLeft += Number(topicOffset.offset) - Number(offset.offset)
+
+        // Handle -1 offsets (no committed offset)
+        if (offset.offset === '-1') {
+          // No committed offset means no messages consumed from this partition
+          consumedMessages += 0
+          // Unconsumed is the total messages in the partition
+          totalLeft += Number(topicOffset.offset) - Number(topicOffset.low)
+        } else {
+          // Consumed messages is the consumer group's offset
+          consumedMessages += Number(offset.offset)
+          // Unconsumed is the difference
+          totalLeft += Number(topicOffset.offset) - Number(offset.offset)
+        }
       }
     }
 

services/apps/integration_run_worker/src/queue/index.ts

@@ -44,7 +44,7 @@ export class WorkerQueueReceiver extends PrioritizedQueueReciever {
 
   override async processMessage(message: IQueueMessage): Promise<void> {
     try {
-      this.log.trace({ messageType: message.type }, 'Processing message!')
+      this.log.info({ messageType: message.type }, 'Processing message!')
 
       const service = new IntegrationRunService(
         this.redisClient,

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

@@ -55,32 +55,68 @@ export class KafkaQueueService extends LoggerBase implements IQueue {
   }
 
   async getQueueMessageCount(conf: IKafkaChannelConfig): Promise<number> {
-    const groupId = conf.name
     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.offset !== offset.offset) {
-          totalLeft += Number(topicOffset.offset) - Number(offset.offset)
+        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
     } catch (err) {
-      this.log.error(err, 'Failed to get message count!')
+      this.log.error({ topic, groupId, err }, 'Failed to get message count!')
       throw err
     } finally {
       await admin.disconnect()