Add helper methods to KafkaTelemetry for interceptor config

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

Author: Copilot

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

@@ -35,32 +35,34 @@ 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
+##### Recommended approach: Configuring interceptors with KafkaTelemetry
 
-The recommended way to use interceptors is to configure them with an `OpenTelemetry` instance.
+The recommended way to use interceptors is to configure them with a `KafkaTelemetry` instance.
 Interceptors will use system properties for additional configuration like captured headers and receive telemetry settings.
 
 For the producer:
 
 ```java
+KafkaTelemetry telemetry = KafkaTelemetry.create(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.put(TracingProducerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
-    new io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetrySupplier(openTelemetry));
+props.putAll(telemetry.producerInterceptorConfigProperties());
 
 Producer<String, String> producer = new KafkaProducer<>(props);
 ```
 
 For the consumer:
 
 ```java
+KafkaTelemetry telemetry = KafkaTelemetry.create(openTelemetry);
+
 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));
+props.putAll(telemetry.consumerInterceptorConfigProperties());
 
 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

@@ -207,6 +207,58 @@ <K, V> ConsumerRecords<K, V> addTracing(
     return Collections.unmodifiableMap(config);
   }
 
+  /**
+   * Returns configuration properties that can be used to enable tracing via {@code
+   * TracingProducerInterceptor}. Add these resulting properties to the configuration map used to
+   * initialize a {@link org.apache.kafka.clients.producer.KafkaProducer}.
+   *
+   * <p>Example usage:
+   *
+   * <pre>{@code
+   * //    KafkaTelemetry telemetry = KafkaTelemetry.create(openTelemetry);
+   * //    Map<String, Object> config = new HashMap<>();
+   * //    config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
+   * //    config.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
+   * //    config.putAll(telemetry.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(
+        TracingProducerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
+        new OpenTelemetrySupplier(openTelemetry));
+    return Collections.unmodifiableMap(config);
+  }
+
+  /**
+   * Returns configuration properties that can be used to enable tracing via {@code
+   * TracingConsumerInterceptor}. Add these resulting properties to the configuration map used to
+   * initialize a {@link org.apache.kafka.clients.consumer.KafkaConsumer}.
+   *
+   * <p>Example usage:
+   *
+   * <pre>{@code
+   * //    KafkaTelemetry telemetry = KafkaTelemetry.create(openTelemetry);
+   * //    Map<String, Object> config = new HashMap<>();
+   * //    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, ...);
+   * //    config.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
+   * //    config.putAll(telemetry.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(
+        TracingConsumerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
+        new OpenTelemetrySupplier(openTelemetry));
+    return Collections.unmodifiableMap(config);
+  }
+
   /**
    * Build and inject span into record.
    *

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

@@ -8,7 +8,6 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.KafkaClientBaseTest;
-import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetrySupplier;
 import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
 import io.opentelemetry.instrumentation.testing.junit.LibraryInstrumentationExtension;
 import java.nio.charset.StandardCharsets;
@@ -29,14 +28,15 @@ abstract class AbstractInterceptorsTest extends KafkaClientBaseTest {
 
   static final String greeting = "Hello Kafka!";
 
+  private static final KafkaTelemetry kafkaTelemetry =
+      KafkaTelemetry.create(testing.getOpenTelemetry());
+
   @Override
   public Map<String, Object> producerProps() {
     Map<String, Object> props = super.producerProps();
     props.put(
         ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingProducerInterceptor.class.getName());
-    props.put(
-        TracingProducerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
-        new OpenTelemetrySupplier(testing.getOpenTelemetry()));
+    props.putAll(kafkaTelemetry.producerInterceptorConfigProperties());
     return props;
   }
 
@@ -45,9 +45,7 @@ public Map<String, Object> consumerProps() {
     Map<String, Object> props = super.consumerProps();
     props.put(
         ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, TracingConsumerInterceptor.class.getName());
-    props.put(
-        TracingConsumerInterceptor.CONFIG_KEY_OPENTELEMETRY_SUPPLIER,
-        new OpenTelemetrySupplier(testing.getOpenTelemetry()));
+    props.putAll(kafkaTelemetry.consumerInterceptorConfigProperties());
     return props;
   }