Avoid changing KafkaTelemetry - move config keys to interceptors

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

Author: Copilot

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

@@ -51,7 +51,8 @@ KafkaTelemetry telemetry = KafkaTelemetry.builder(openTelemetry)
 Map<String, Object> props = new HashMap<>();
 props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
 props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
-props.putAll(telemetry.producerInterceptorConfigProperties());
+props.put(TracingProducerInterceptor.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER,
+    new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaTelemetrySupplier(telemetry));
 
 Producer<String, String> producer = new KafkaProducer<>(props);
 ```
@@ -68,7 +69,8 @@ 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.putAll(telemetry.consumerInterceptorConfigProperties());
+props.put(TracingConsumerInterceptor.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER,
+    new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaTelemetrySupplier(telemetry));
 
 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/KafkaTelemetry.java

@@ -21,7 +21,6 @@
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaProcessRequest;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaProducerRequest;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaReceiveRequest;
-import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaTelemetrySupplier;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaUtil;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.MetricsReporterList;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetryMetricsReporter;
@@ -208,61 +207,6 @@ <K, V> ConsumerRecords<K, V> addTracing(
     return Collections.unmodifiableMap(config);
   }
 
-  /**
-   * Returns a config property key that can be used to pass this {@link KafkaTelemetry} instance to
-   * interceptors.
-   *
-   * <p>This is an internal config key used by the library instrumentation interceptors.
-   */
-  static final String CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER =
-      "opentelemetry.experimental.kafka-telemetry.supplier";
-
-  /**
-   * Produces a set of kafka producer config properties to configure tracing in {@code
-   * TracingProducerInterceptor}. Add these resulting properties to the configuration map used to
-   * initialize a {@link KafkaProducer}.
-   *
-   * <p>Example usage:
-   *
-   * <pre>{@code
-   * //    Map<String, Object> config = new HashMap<>();
-   * //    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
-   * //    config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
-   * //    config.putAll(kafkaTelemetry.producerInterceptorConfigProperties());
-   * //    try (KafkaProducer<?, ?> producer = new KafkaProducer<>(config)) { ... }
-   * }</pre>
-   *
-   * @return the kafka producer interceptor config properties
-   */
-  public Map<String, ?> producerInterceptorConfigProperties() {
-    Map<String, Object> config = new HashMap<>();
-    config.put(CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER, new KafkaTelemetrySupplier(this));
-    return Collections.unmodifiableMap(config);
-  }
-
-  /**
-   * Produces a set of kafka consumer config properties to configure tracing in {@code
-   * TracingConsumerInterceptor}. Add these resulting properties to the configuration map used to
-   * initialize a {@link KafkaConsumer}.
-   *
-   * <p>Example usage:
-   *
-   * <pre>{@code
-   * //    Map<String, Object> config = new HashMap<>();
-   * //    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
-   * //    config.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
-   * //    config.putAll(kafkaTelemetry.consumerInterceptorConfigProperties());
-   * //    try (KafkaConsumer<?, ?> consumer = new KafkaConsumer<>(config)) { ... }
-   * }</pre>
-   *
-   * @return the kafka consumer interceptor config properties
-   */
-  public Map<String, ?> consumerInterceptorConfigProperties() {
-    Map<String, Object> config = new HashMap<>();
-    config.put(CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER, new KafkaTelemetrySupplier(this));
-    return Collections.unmodifiableMap(config);
-  }
-
   /**
    * Build and inject span into record.
    *

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

@@ -31,6 +31,9 @@
  */
 public class TracingConsumerInterceptor<K, V> implements ConsumerInterceptor<K, V> {
 
+  public static final String CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER =
+      "opentelemetry.experimental.kafka-telemetry.supplier";
+
   @Nullable private KafkaTelemetry telemetry;
   private String consumerGroup;
   private String clientId;
@@ -63,7 +66,7 @@ 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(KafkaTelemetry.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER);
+    Object telemetrySupplier = configs.get(CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER);
     if (telemetrySupplier == null) {
       // Fallback to GlobalOpenTelemetry if not configured
       this.telemetry =
@@ -82,15 +85,15 @@ public void configure(Map<String, ?> configs) {
     if (!(telemetrySupplier instanceof Supplier)) {
       throw new IllegalStateException(
           "Configuration property "
-              + KafkaTelemetry.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
+              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
               + " is not instance of Supplier");
     }
 
     Object kafkaTelemetry = ((Supplier<?>) telemetrySupplier).get();
     if (!(kafkaTelemetry instanceof KafkaTelemetry)) {
       throw new IllegalStateException(
           "Configuration property "
-              + KafkaTelemetry.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
+              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
               + " supplier does not return KafkaTelemetry instance");
     }
 

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

@@ -26,6 +26,9 @@
  */
 public class TracingProducerInterceptor<K, V> implements ProducerInterceptor<K, V> {
 
+  public static final String CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER =
+      "opentelemetry.experimental.kafka-telemetry.supplier";
+
   @Nullable private KafkaTelemetry telemetry;
   @Nullable private String clientId;
 
@@ -48,7 +51,7 @@ public void close() {}
   public void configure(Map<String, ?> configs) {
     clientId = Objects.toString(configs.get(ProducerConfig.CLIENT_ID_CONFIG), null);
 
-    Object telemetrySupplier = configs.get(KafkaTelemetry.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER);
+    Object telemetrySupplier = configs.get(CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER);
     if (telemetrySupplier == null) {
       // Fallback to GlobalOpenTelemetry if not configured
       this.telemetry =
@@ -63,15 +66,15 @@ public void configure(Map<String, ?> configs) {
     if (!(telemetrySupplier instanceof Supplier)) {
       throw new IllegalStateException(
           "Configuration property "
-              + KafkaTelemetry.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
+              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
               + " is not instance of Supplier");
     }
 
     Object kafkaTelemetry = ((Supplier<?>) telemetrySupplier).get();
     if (!(kafkaTelemetry instanceof KafkaTelemetry)) {
       throw new IllegalStateException(
           "Configuration property "
-              + KafkaTelemetry.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
+              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
               + " supplier does not return KafkaTelemetry instance");
     }
 

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

@@ -45,7 +45,11 @@ public Map<String, Object> producerProps() {
     Map<String, Object> props = super.producerProps();
     props.put(
         ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
-    props.putAll(kafkaTelemetry.producerInterceptorConfigProperties());
+    props.put(
+        TracingProducerInterceptor.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER,
+        new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal
+                .KafkaTelemetrySupplier(
+            kafkaTelemetry));
     return props;
   }
 
@@ -57,7 +61,11 @@ public Map<String, Object> consumerProps() {
     Map<String, Object> props = super.consumerProps();
     props.put(
         ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
-    props.putAll(kafkaTelemetry.consumerInterceptorConfigProperties());
+    props.put(
+        TracingConsumerInterceptor.CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER,
+        new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal
+                .KafkaTelemetrySupplier(
+            kafkaTelemetry));
     return props;
   }