Add kafka client metrics

instrumentation/kafka/kafka-clients/kafka-clients-0.11/testing/src/main/groovy/io/opentelemetry/instrumentation/kafkaclients/KafkaClientBaseTest.groovy

@@ -6,6 +6,8 @@
 package io.opentelemetry.instrumentation.kafkaclients
 
 import io.opentelemetry.instrumentation.test.InstrumentationSpecification
+import org.testcontainers.utility.DockerImageName
+
 import java.time.Duration
 import org.apache.kafka.clients.admin.AdminClient
 import org.apache.kafka.clients.admin.NewTopic
@@ -46,7 +48,7 @@ abstract class KafkaClientBaseTest extends InstrumentationSpecification {
   static TopicPartition topicPartition = new TopicPartition(SHARED_TOPIC, 0)
 
   def setupSpec() {
-    kafka = new KafkaContainer()
+    kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.4.3"))
       .withLogConsumer(new Slf4jLogConsumer(logger))
       .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
       .withStartupTimeout(Duration.ofMinutes(1))

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/build.gradle.kts

@@ -10,6 +10,10 @@ dependencies {
   testImplementation("com.fasterxml.jackson.core:jackson-databind:2.10.2")
 
   testImplementation("org.testcontainers:kafka")
+  testImplementation("org.testcontainers:junit-jupiter")
+
+  testCompileOnly("com.google.auto.value:auto-value-annotations")
+  testAnnotationProcessor("com.google.auto.value:auto-value")
 }
 
 tasks {

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/KafkaTelemetry.java

@@ -17,17 +17,25 @@
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
 import io.opentelemetry.instrumentation.kafka.internal.KafkaConsumerRecordGetter;
 import io.opentelemetry.instrumentation.kafka.internal.KafkaHeadersSetter;
+import io.opentelemetry.instrumentation.kafka.internal.OpenTelemetryMetricsReporter;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.Future;
 import java.util.function.BiFunction;
 import java.util.logging.Logger;
+import org.apache.kafka.clients.CommonClientConfigs;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.Callback;
+import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.metrics.MetricsReporter;
 
 public final class KafkaTelemetry {
   private static final Logger logger = Logger.getLogger(KafkaTelemetry.class.getName());
@@ -76,6 +84,41 @@ public <K, V> Consumer<K, V> wrap(Consumer<K, V> consumer) {
     return new TracingConsumer<>(consumer, this);
   }
 
+  /**
+   * Produces a set of kafka client config properties (consumer or producer) to register a {@link
+   * MetricsReporter} that records metrics to an {@code openTelemetry} instance. Add these resulting
+   * properties to the configuration map used to initialize a {@link KafkaConsumer} or {@link
+   * KafkaProducer}.
+   *
+   * <p>For producers:
+   *
+   * <pre>{@code
+   * //    Map<String, Object> config = new HashMap<>();
+   * //    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
+   * //    config.putAll(kafkaTelemetry.metricConfigProperties());
+   * //    try (KafkaProducer<?, ?> producer = new KafkaProducer<>(config)) { ... }
+   * }</pre>
+   *
+   * <p>For consumers:
+   *
+   * <pre>{@code
+   * //    Map<String, Object> config = new HashMap<>();
+   * //    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
+   * //    config.putAll(kafkaTelemetry.metricConfigProperties());
+   * //    try (KafkaConsumer<?, ?> consumer = new KafkaConsumer<>(config)) { ... }
+   * }</pre>
+   *
+   * @return the kafka client properties
+   */
+  public Map<String, ?> metricConfigProperties() {
+    Map<String, Object> config = new HashMap<>();
+    config.put(
+        CommonClientConfigs.METRIC_REPORTER_CLASSES_CONFIG,
+        OpenTelemetryMetricsReporter.class.getName());
+    config.put(OpenTelemetryMetricsReporter.CONFIG_KEY_OPENTELEMETRY_INSTANCE, openTelemetry);
+    return Collections.unmodifiableMap(config);
+  }
+
   /**
    * Build and inject span into record.
    *