Add overall consumer poll rate.

linkedin · Feb 7, 2025 · d525264 · d525264
1 parent 010a17c
commit d525264
Showing 1 changed file with 7 additions and 2 deletions.
diff --git a/...ts/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/ConsumptionTask.java b/...ts/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/ConsumptionTask.java
@@ -66,6 +66,8 @@ class ConsumptionTask implements Runnable {
   private final Map<PubSubTopicPartition, Rate> messageRatePerTopicPartition = new VeniceConcurrentHashMap<>();
   private final Map<PubSubTopicPartition, Rate> bytesRatePerTopicPartition = new VeniceConcurrentHashMap<>();
   private final Map<PubSubTopicPartition, Rate> pollRatePerTopicPartition = new VeniceConcurrentHashMap<>();
+
+  private final Rate overallConsumerPollRate = createRate(System.currentTimeMillis());
   private final RedundantExceptionFilter redundantExceptionFilter;
   private final Map<PubSubTopicPartition, Long> lastSuccessfulPollTimestampPerTopicPartition =
       new VeniceConcurrentHashMap<>();
@@ -196,6 +198,7 @@ public void run() {
               consumedDataReceiver.write(topicPartitionMessages);
               checkSlowPartitionWithHighLag(pubSubTopicPartition);
             }
+            overallConsumerPollRate.record(1, lastSuccessfulPollTimestamp);
             aggStats.recordTotalConsumerRecordsProducingToWriterBufferLatency(
                 LatencyUtils.getElapsedTimeFromMsToMs(beforeProducingToWriteBufferTimestamp));
             aggStats.recordTotalNonZeroPollResultNum(polledPubSubMessagesCount);
@@ -302,15 +305,17 @@ private void checkSlowPartitionWithHighLag(PubSubTopicPartition pubSubTopicParti
     Long offsetLag = offsetLagGetter.apply(pubSubTopicPartition);
     Double messageRate = getMessageRate(pubSubTopicPartition);
     Double pollRate = getPollRate(pubSubTopicPartition);
+    Double consumerPollRate = overallConsumerPollRate.measure(metricConfig, System.currentTimeMillis());
     String slowTaskWithPartitionStr = consumptionTaskIdStr + " - " + pubSubTopicPartition;
     if (offsetLag > 200000 && messageRate < 200
         && !redundantExceptionFilter.isRedundantException(slowTaskWithPartitionStr)) {
       LOGGER.warn(
-          "Slow partition with high lag detected: {}. Lag: {}, Message Rate: {}, Poll Rate: {}",
+          "Slow partition with high lag detected: {}. Lag: {}, Message Rate: {}, Poll Rate: {}，Consumer Poll Rate: {}",
           pubSubTopicPartition,
           offsetLag,
           messageRate,
-          pollRate);
+          pollRate,
+          consumerPollRate);
     }
   }