feat: Add kafka/consumer/pollIdleRatio metric. (apache#19366)

The metric corresponds to the Kafka consumer metric "poll-idle-ratio-avg".

Author: gianm

extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json

@@ -97,6 +97,7 @@
   "kafka/consumer/recordsPerRequestAvg" : { "dimensions" : ["topic"], "type" : "gauge", "help": "Average records per fetch request as seen by the consumer of a Kafka indexing task (per topic)."},
   "kafka/consumer/outgoingBytes" : { "dimensions" : ["node_id"], "type" : "count", "help": "Bytes sent to Kafka brokers by the consumer of a Kafka indexing task (per node)."},
   "kafka/consumer/incomingBytes" : { "dimensions" : ["node_id"], "type" : "count", "help": "Bytes received from Kafka brokers by the consumer of a Kafka indexing task (per node)."},
+  "kafka/consumer/pollIdleRatio" : { "dimensions" : [], "type" : "gauge", "help": "Average fraction of time the consumer of a Kafka indexing task spent idle (not in poll). 0 means never idle, 1 means always idle."},
 
   "ingest/count" : { "dimensions" : ["dataSource", "taskType"], "type" : "count", "help": "Count of 1 every time an ingestion job runs (includes compaction jobs). Aggregate using dimensions." },
   "ingest/segments/count" : { "dimensions" : ["dataSource", "taskType"], "type" : "count", "help": "Count of final segments created by job (includes tombstones)." },

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaConsumerMonitor.java

@@ -127,6 +127,12 @@ public class KafkaConsumerMonitor extends AbstractMonitor
               "kafka/consumer/incomingBytes",
               Set.of(CLIENT_ID_TAG, NODE_ID_TAG),
               KafkaConsumerMetric.MetricType.COUNTER
+          ),
+          new KafkaConsumerMetric(
+              POLL_IDLE_RATIO_METRIC_NAME,
+              "kafka/consumer/pollIdleRatio",
+              Set.of(CLIENT_ID_TAG),
+              KafkaConsumerMetric.MetricType.GAUGE
           )
       ).collect(Collectors.toMap(KafkaConsumerMetric::getKafkaMetricName, Function.identity()));
 

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaRecordSupplierTest.java

@@ -484,6 +484,7 @@ public void testPoll() throws InterruptedException, ExecutionException
     emitter.verifyEmitted("kafka/consumer/recordsPerRequestAvg", 1);
     emitter.verifyEmitted("kafka/consumer/incomingBytes", 2);
     emitter.verifyEmitted("kafka/consumer/outgoingBytes", 2);
+    emitter.verifyEmitted("kafka/consumer/pollIdleRatio", 1);
 
     recordSupplier.close();
     Assert.assertFalse(monitor.monitor(emitter));

processing/src/main/resources/loggingEmitterAllowedMetrics.json

@@ -76,6 +76,7 @@
     "kafka/consumer/fetchSizeMax": [],
     "kafka/consumer/incomingBytes": [],
     "kafka/consumer/outgoingBytes": [],
+    "kafka/consumer/pollIdleRatio": [],
     "kafka/consumer/recordsConsumed": [],
     "kafka/consumer/recordsLag": [],
     "kafka/consumer/recordsPerRequestAvg": [],

processing/src/test/resources/loggingEmitterAllowedMetrics.json

@@ -76,6 +76,7 @@
   "kafka/consumer/fetchSizeMax": [],
   "kafka/consumer/incomingBytes": [],
   "kafka/consumer/outgoingBytes": [],
+  "kafka/consumer/pollIdleRatio": [],
   "kafka/consumer/recordsConsumed": [],
   "kafka/consumer/recordsLag": [],
   "kafka/consumer/recordsPerRequestAvg": [],