Use OpenTelemetrySupplier instead of KafkaTelemetrySupplier

Co-authored-by: trask <[email protected]>

Author: Copilot

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

@@ -35,42 +35,32 @@ 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 KafkaTelemetry
+##### Recommended approach: Configuring interceptors with OpenTelemetry
 
-The recommended way to use interceptors is to configure them with a `KafkaTelemetry` instance.
-This gives you full control over the configuration, including which `OpenTelemetry` instance to use,
-whether to enable receive telemetry, and which headers to capture.
+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, configure the interceptor with the KafkaTelemetry instance:
+For the producer:
 
 ```java
-KafkaTelemetry telemetry = KafkaTelemetry.builder(openTelemetry)
-    .setCapturedHeaders(Arrays.asList("custom-header"))
-    .build();
-
 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_KAFKA_TELEMETRY_SUPPLIER,
-    new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaTelemetrySupplier(telemetry));
+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, configure the interceptor with the KafkaTelemetry instance:
+For the consumer:
 
 ```java
-KafkaTelemetry telemetry = KafkaTelemetry.builder(openTelemetry)
-    .setMessagingReceiveInstrumentationEnabled(true)
-    .setCapturedHeaders(Arrays.asList("custom-header"))
-    .build();
-
 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_KAFKA_TELEMETRY_SUPPLIER,
-    new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaTelemetrySupplier(telemetry));
+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);
 ```

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

@@ -9,14 +9,15 @@
 
 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 java.util.function.Supplier;
 import javax.annotation.Nullable;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerInterceptor;
@@ -31,8 +32,7 @@
  */
 public class TracingConsumerInterceptor<K, V> implements ConsumerInterceptor<K, V> {
 
-  public static final String CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER =
-      "opentelemetry.experimental.kafka-telemetry.supplier";
+  public static final String CONFIG_KEY_OPENTELEMETRY_SUPPLIER = "opentelemetry.supplier";
 
   @Nullable private KafkaTelemetry telemetry;
   private String consumerGroup;
@@ -66,37 +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);
 
-    Object telemetrySupplier = configs.get(CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER);
-    if (telemetrySupplier == null) {
+    OpenTelemetry openTelemetry;
+    Object openTelemetrySupplier = configs.get(CONFIG_KEY_OPENTELEMETRY_SUPPLIER);
+    if (openTelemetrySupplier == null) {
       // Fallback to GlobalOpenTelemetry if not configured
-      this.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();
-      return;
+      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();
     }
 
-    if (!(telemetrySupplier instanceof Supplier)) {
-      throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " is not instance of Supplier");
-    }
-
-    Object kafkaTelemetry = ((Supplier<?>) telemetrySupplier).get();
-    if (!(kafkaTelemetry instanceof KafkaTelemetry)) {
-      throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " supplier does not return KafkaTelemetry instance");
-    }
-
-    this.telemetry = (KafkaTelemetry) kafkaTelemetry;
+    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,10 +9,11 @@
 
 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 java.util.function.Supplier;
 import javax.annotation.Nullable;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerInterceptor;
@@ -26,8 +27,7 @@
  */
 public class TracingProducerInterceptor<K, V> implements ProducerInterceptor<K, V> {
 
-  public static final String CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER =
-      "opentelemetry.experimental.kafka-telemetry.supplier";
+  public static final String CONFIG_KEY_OPENTELEMETRY_SUPPLIER = "opentelemetry.supplier";
 
   @Nullable private KafkaTelemetry telemetry;
   @Nullable private String clientId;
@@ -51,33 +51,26 @@ public void close() {}
   public void configure(Map<String, ?> configs) {
     clientId = Objects.toString(configs.get(ProducerConfig.CLIENT_ID_CONFIG), null);
 
-    Object telemetrySupplier = configs.get(CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER);
-    if (telemetrySupplier == null) {
+    OpenTelemetry openTelemetry;
+    Object openTelemetrySupplier = configs.get(CONFIG_KEY_OPENTELEMETRY_SUPPLIER);
+    if (openTelemetrySupplier == null) {
       // Fallback to GlobalOpenTelemetry if not configured
-      this.telemetry =
-          KafkaTelemetry.builder(GlobalOpenTelemetry.get())
-              .setCapturedHeaders(
-                  ConfigPropertiesUtil.getList(
-                      "otel.instrumentation.messaging.experimental.capture-headers", emptyList()))
-              .build();
-      return;
+      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();
     }
 
-    if (!(telemetrySupplier instanceof Supplier)) {
-      throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " is not instance of Supplier");
-    }
-
-    Object kafkaTelemetry = ((Supplier<?>) telemetrySupplier).get();
-    if (!(kafkaTelemetry instanceof KafkaTelemetry)) {
-      throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " supplier does not return KafkaTelemetry instance");
-    }
-
-    this.telemetry = (KafkaTelemetry) kafkaTelemetry;
+    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

@@ -46,10 +46,10 @@ public Map<String, Object> producerProps() {
     props.put(
         ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
     props.put(
-        TracingProducerInterceptor.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER,
+        TracingProducerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
         new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal
-                .KafkaTelemetrySupplier(
-            kafkaTelemetry));
+                .OpenTelemetrySupplier(
+            testing.getOpenTelemetry()));
     return props;
   }
 
@@ -62,10 +62,10 @@ public Map<String, Object> consumerProps() {
     props.put(
         ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
     props.put(
-        TracingConsumerInterceptor.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER,
+        TracingConsumerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
         new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal
-                .KafkaTelemetrySupplier(
-            kafkaTelemetry));
+                .OpenTelemetrySupplier(
+            testing.getOpenTelemetry()));
     return props;
   }