Add kafka.request.time.avg (#1135)

Author: mackjmr

jmx-metrics/docs/target-systems/kafka.md

@@ -42,6 +42,12 @@ These metrics are sourced from Kafka's exposed Yammer metrics for each instance:
 * Attributes: `type`
 * Instrument Type: DoubleValueObserver
 
+* Name: `kafka.request.time.avg`
+* Description: The average time the broker has taken to service requests
+* Unit: `ms`
+* Attributes: `type`
+* Instrument Type: DoubleValueObserver
+
 * Name: `kafka.network.io`
 * Description: The bytes received or sent by the broker
 * Unit: `by`

jmx-metrics/src/integrationTest/java/io/opentelemetry/contrib/jmxmetrics/target_systems/KafkaIntegrationTest.java

@@ -126,6 +126,15 @@ List<Consumer<Metric>> kafkaBrokerAssertions() {
                 attrs -> attrs.containsOnly(entry("type", "produce")),
                 attrs -> attrs.containsOnly(entry("type", "fetchfollower")),
                 attrs -> attrs.containsOnly(entry("type", "fetchconsumer"))),
+        metric ->
+            assertGaugeWithAttributes(
+                metric,
+                "kafka.request.time.avg",
+                "The average time the broker has taken to service requests",
+                "ms",
+                attrs -> attrs.containsOnly(entry("type", "produce")),
+                attrs -> attrs.containsOnly(entry("type", "fetchfollower")),
+                attrs -> attrs.containsOnly(entry("type", "fetchconsumer"))),
         metric ->
             assertGauge(
                 metric,

jmx-metrics/src/main/resources/target-systems/kafka.groovy

@@ -86,8 +86,14 @@ otel.instrument(requestTime,
     "type" : { mbean -> mbean.name().getKeyProperty("request").toLowerCase() },
   ],
   "99thPercentile", otel.&doubleValueCallback)
-
-
+otel.instrument(requestTime,
+  "kafka.request.time.avg",
+  "The average time the broker has taken to service requests",
+  "ms",
+  [
+    "type" : { mbean -> mbean.name().getKeyProperty("request").toLowerCase() },
+  ],
+  "Mean", otel.&doubleValueCallback)
 
 def network = otel.mbeans(["kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec",
                           "kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec"])