fix(plugin-server): update kafkaAbsolutePartitionCount even on empty … (

#17791) fix(plugin-server): update kafkaAbsolutePartitionCount even on empty consume loop
PostHog · Oct 5, 2023 · 0d6218f · 0d6218f
1 parent bb996d1
commit 0d6218f
Showing 1 changed file with 4 additions and 4 deletions.
diff --git a/plugin-server/src/kafka/batch-consumer.ts b/plugin-server/src/kafka/batch-consumer.ts
@@ -181,15 +181,15 @@ export const startBatchConsumer = async ({
                 // protocol. If we never manage to consume, we don't want our health checks to pass.
                 lastConsumeTime = Date.now()
 
+                for (const [topic, count] of countPartitionsPerTopic(consumer.assignments())) {
+                    kafkaAbsolutePartitionCount.labels({ topic }).set(count)
+                }
+
                 if (!messages) {
                     status.debug('🔁', 'main_loop_empty_batch', { cause: 'undefined' })
                     continue
                 }
 
-                for (const [topic, count] of countPartitionsPerTopic(consumer.assignments())) {
-                    kafkaAbsolutePartitionCount.labels({ topic }).set(count)
-                }
-
                 status.debug('🔁', 'main_loop_consumed', { messagesLength: messages.length })
                 if (!messages.length) {
                     status.debug('🔁', 'main_loop_empty_batch', { cause: 'empty' })