Add configurable OpenTelemetry kafka-clients interceptors

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/README.md

@@ -35,6 +35,41 @@ The Kafka clients API provides a way to "intercept" messages before they are sen
 The OpenTelemetry instrumented Kafka library provides two interceptors to be configured to add tracing information automatically.
 The interceptor class has to be set in the properties bag used to create the Kafka client.
 
+##### Recommended approach: Configuring interceptors with OpenTelemetry
+
+The recommended way to use interceptors is to configure them with an `OpenTelemetry` instance.
+Interceptors will use system properties for additional configuration like captured headers and receive telemetry settings.
+
+For the producer:
+
+```java
+Map<String, Object> props = new HashMap<>();
+props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
+props.put(TracingProducerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
+    new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetrySupplier(openTelemetry));
+
+Producer<String, String> producer = new KafkaProducer<>(props);
+```
+
+For the consumer:
+
+```java
+Map<String, Object> props = new HashMap<>();
+props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
+props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
+props.put(TracingConsumerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
+    new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetrySupplier(openTelemetry));
+
+Consumer<String, String> consumer = new KafkaConsumer<>(props);
+```
+
+##### Alternative: Using interceptors with global OpenTelemetry
+
+If you don't explicitly configure the interceptors with a `KafkaTelemetry` instance, they will fall back to using
+`GlobalOpenTelemetry.get()` and system properties for configuration.
+
 Use the `TracingProducerInterceptor` for the producer in order to create a "send" span automatically, each time a message is sent.
 
 ```java
@@ -47,6 +82,10 @@ Use the `TracingConsumerInterceptor` for the consumer in order to create a "rece
 props.setProperty(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
 ```
 
+The interceptors will use the following system properties for configuration:
+- `otel.instrumentation.messaging.experimental.receive-telemetry.enabled` - Enable receive telemetry (default: false)
+- `otel.instrumentation.messaging.experimental.capture-headers` - List of headers to capture as span attributes
+
 #### Wrapping clients
 
 The other way is by wrapping the Kafka client with a tracing enabled Kafka client.

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

@@ -9,13 +9,16 @@
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.instrumentation.api.internal.ConfigPropertiesUtil;
 import io.opentelemetry.instrumentation.api.internal.Timer;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaConsumerContext;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaConsumerContextUtil;
+import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetrySupplier;
 import java.util.Map;
 import java.util.Objects;
+import javax.annotation.Nullable;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerInterceptor;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -29,22 +32,18 @@
  */
 public class TracingConsumerInterceptor<K, V> implements ConsumerInterceptor<K, V> {
 
-  private static final KafkaTelemetry telemetry =
-      KafkaTelemetry.builder(GlobalOpenTelemetry.get())
-          .setMessagingReceiveInstrumentationEnabled(
-              ConfigPropertiesUtil.getBoolean(
-                  "otel.instrumentation.messaging.experimental.receive-telemetry.enabled", false))
-          .setCapturedHeaders(
-              ConfigPropertiesUtil.getList(
-                  "otel.instrumentation.messaging.experimental.capture-headers", emptyList()))
-          .build();
+  public static final String CONFIG_KEY_OPENTELEMETRY_SUPPLIER = "opentelemetry.supplier";
 
+  @Nullable private KafkaTelemetry telemetry;
   private String consumerGroup;
   private String clientId;
 
   @Override
   @CanIgnoreReturnValue
   public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
+    if (telemetry == null) {
+      return records;
+    }
     // timer should be started before fetching ConsumerRecords, but there is no callback for that
     Timer timer = Timer.start();
     Context receiveContext = telemetry.buildAndFinishSpan(records, consumerGroup, clientId, timer);
@@ -67,6 +66,29 @@ public void configure(Map<String, ?> configs) {
     consumerGroup = Objects.toString(configs.get(ConsumerConfig.GROUP_ID_CONFIG), null);
     clientId = Objects.toString(configs.get(ConsumerConfig.CLIENT_ID_CONFIG), null);
 
-    // TODO: support experimental attributes config
+    OpenTelemetry openTelemetry;
+    Object openTelemetrySupplier = configs.get(CONFIG_KEY_OPENTELEMETRY_SUPPLIER);
+    if (openTelemetrySupplier == null) {
+      // Fallback to GlobalOpenTelemetry if not configured
+      openTelemetry = GlobalOpenTelemetry.get();
+    } else {
+      if (!(openTelemetrySupplier instanceof OpenTelemetrySupplier)) {
+        throw new IllegalStateException(
+            "Configuration property "
+                + CONFIG_KEY_OPENTELEMETRY_SUPPLIER
+                + " is not instance of OpenTelemetrySupplier");
+      }
+      openTelemetry = ((OpenTelemetrySupplier) openTelemetrySupplier).get();
+    }
+
+    this.telemetry =
+        KafkaTelemetry.builder(openTelemetry)
+            .setMessagingReceiveInstrumentationEnabled(
+                ConfigPropertiesUtil.getBoolean(
+                    "otel.instrumentation.messaging.experimental.receive-telemetry.enabled", false))
+            .setCapturedHeaders(
+                ConfigPropertiesUtil.getList(
+                    "otel.instrumentation.messaging.experimental.capture-headers", emptyList()))
+            .build();
   }
 }

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

@@ -9,7 +9,9 @@
 
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.instrumentation.api.internal.ConfigPropertiesUtil;
+import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetrySupplier;
 import java.util.Map;
 import java.util.Objects;
 import javax.annotation.Nullable;
@@ -25,19 +27,17 @@
  */
 public class TracingProducerInterceptor<K, V> implements ProducerInterceptor<K, V> {
 
-  private static final KafkaTelemetry telemetry =
-      KafkaTelemetry.builder(GlobalOpenTelemetry.get())
-          .setCapturedHeaders(
-              ConfigPropertiesUtil.getList(
-                  "otel.instrumentation.messaging.experimental.capture-headers", emptyList()))
-          .build();
+  public static final String CONFIG_KEY_OPENTELEMETRY_SUPPLIER = "opentelemetry.supplier";
 
+  @Nullable private KafkaTelemetry telemetry;
   @Nullable private String clientId;
 
   @Override
   @CanIgnoreReturnValue
   public ProducerRecord<K, V> onSend(ProducerRecord<K, V> producerRecord) {
-    telemetry.buildAndInjectSpan(producerRecord, clientId);
+    if (telemetry != null) {
+      telemetry.buildAndInjectSpan(producerRecord, clientId);
+    }
     return producerRecord;
   }
 
@@ -48,9 +48,29 @@ public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {}
   public void close() {}
 
   @Override
-  public void configure(Map<String, ?> map) {
-    clientId = Objects.toString(map.get(ProducerConfig.CLIENT_ID_CONFIG), null);
+  public void configure(Map<String, ?> configs) {
+    clientId = Objects.toString(configs.get(ProducerConfig.CLIENT_ID_CONFIG), null);
 
-    // TODO: support experimental attributes config
+    OpenTelemetry openTelemetry;
+    Object openTelemetrySupplier = configs.get(CONFIG_KEY_OPENTELEMETRY_SUPPLIER);
+    if (openTelemetrySupplier == null) {
+      // Fallback to GlobalOpenTelemetry if not configured
+      openTelemetry = GlobalOpenTelemetry.get();
+    } else {
+      if (!(openTelemetrySupplier instanceof OpenTelemetrySupplier)) {
+        throw new IllegalStateException(
+            "Configuration property "
+                + CONFIG_KEY_OPENTELEMETRY_SUPPLIER
+                + " is not instance of OpenTelemetrySupplier");
+      }
+      openTelemetry = ((OpenTelemetrySupplier) openTelemetrySupplier).get();
+    }
+
+    this.telemetry =
+        KafkaTelemetry.builder(openTelemetry)
+            .setCapturedHeaders(
+                ConfigPropertiesUtil.getList(
+                    "otel.instrumentation.messaging.experimental.capture-headers", emptyList()))
+            .build();
   }
 }

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/AbstractInterceptorsTest.java

@@ -28,19 +28,44 @@ abstract class AbstractInterceptorsTest extends KafkaClientBaseTest {
 
   static final String greeting = "Hello Kafka!";
 
+  private static KafkaTelemetry kafkaTelemetry;
+
+  protected static KafkaTelemetry createKafkaTelemetry() {
+    return KafkaTelemetry.builder(testing.getOpenTelemetry())
+        .setCapturedHeaders(java.util.Collections.singletonList("Test-Message-Header"))
+        .setMessagingReceiveInstrumentationEnabled(true)
+        .build();
+  }
+
   @Override
   public Map<String, Object> producerProps() {
+    if (kafkaTelemetry == null) {
+      kafkaTelemetry = createKafkaTelemetry();
+    }
     Map<String, Object> props = super.producerProps();
     props.put(
         ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
+    props.put(
+        TracingProducerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
+        new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal
+                .OpenTelemetrySupplier(
+            testing.getOpenTelemetry()));
     return props;
   }
 
   @Override
   public Map<String, Object> consumerProps() {
+    if (kafkaTelemetry == null) {
+      kafkaTelemetry = createKafkaTelemetry();
+    }
     Map<String, Object> props = super.consumerProps();
     props.put(
         ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
+    props.put(
+        TracingConsumerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
+        new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal
+                .OpenTelemetrySupplier(
+            testing.getOpenTelemetry()));
     return props;
   }
 

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/InterceptorsSuppressReceiveSpansTest.java

@@ -23,6 +23,12 @@
 
 class InterceptorsSuppressReceiveSpansTest extends AbstractInterceptorsTest {
 
+  protected static KafkaTelemetry createKafkaTelemetry() {
+    return KafkaTelemetry.builder(testing.getOpenTelemetry())
+        .setMessagingReceiveInstrumentationEnabled(false)
+        .build();
+  }
+
   @SuppressWarnings("deprecation") // using deprecated semconv
   @Override
   void assertTraces() {