add capture headers to InterceptorsTest

Author: mznet

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

@@ -36,6 +36,7 @@ tasks {
       excludeTestsMatching("WrapperSuppressReceiveSpansTest")
     }
     jvmArgs("-Dotel.instrumentation.messaging.experimental.receive-telemetry.enabled=true")
+    jvmArgs("-Dotel.instrumentation.messaging.experimental.capture-headers=baggage")
   }
 
   check {

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

@@ -18,10 +18,12 @@
 import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.trace.SpanContext;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.sdk.trace.data.LinkData;
 import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
 import java.util.concurrent.atomic.AtomicReference;
 import org.assertj.core.api.AbstractLongAssert;
 import org.assertj.core.api.AbstractStringAssert;
@@ -42,6 +44,11 @@ void assertTraces() {
                       .hasKind(SpanKind.PRODUCER)
                       .hasParent(trace.getSpan(0))
                       .hasAttributesSatisfyingExactly(
+                          equalTo(
+                              AttributeKey.stringArrayKey("messaging.header.baggage"),
+                              Arrays.asList(
+                                  "test-baggage-key-1=test-baggage-value-1",
+                                  "test-baggage-key-2=test-baggage-value-2")),
                           equalTo(MESSAGING_SYSTEM, "kafka"),
                           equalTo(MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
                           equalTo(MESSAGING_OPERATION, "publish"),
@@ -64,6 +71,11 @@ void assertTraces() {
                         .hasNoParent()
                         .hasLinksSatisfying(links -> assertThat(links).isEmpty())
                         .hasAttributesSatisfyingExactly(
+                            equalTo(
+                                AttributeKey.stringArrayKey("messaging.header.baggage"),
+                                Arrays.asList(
+                                    "test-baggage-key-1=test-baggage-value-1",
+                                    "test-baggage-key-2=test-baggage-value-2")),
                             equalTo(MESSAGING_SYSTEM, "kafka"),
                             equalTo(MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
                             equalTo(MESSAGING_OPERATION, "receive"),
@@ -78,6 +90,11 @@ void assertTraces() {
                         .hasParent(trace.getSpan(0))
                         .hasLinks(LinkData.create(producerSpanContext.get()))
                         .hasAttributesSatisfyingExactly(
+                            equalTo(
+                                AttributeKey.stringArrayKey("messaging.header.baggage"),
+                                Arrays.asList(
+                                    "test-baggage-key-1=test-baggage-value-1",
+                                    "test-baggage-key-2=test-baggage-value-2")),
                             equalTo(MESSAGING_SYSTEM, "kafka"),
                             equalTo(MESSAGING_DESTINATION_NAME, SHARED_TOPIC),
                             equalTo(MESSAGING_OPERATION, "process"),

instrumentation/kafka/kafka-clients/kafka-clients-2.6/metadata.yaml

@@ -1,2 +1,11 @@
 description: >
-  This instrumentation provides a library integeration that enables messaging spans and metrics for Apache Kafka 2.6+ clients.
+  This instrumentation provides both library and wrapper integrations that enable messaging spans and metrics for Apache Kafka 2.6+ clients.
+configurations:
+  - name: otel.instrumentation.messaging.experimental.capture-headers
+    description: A comma-separated list of header names to capture as span attributes.
+    type: list
+    default: ''
+  - name: otel.instrumentation.messaging.experimental.receive-telemetry.enabled
+    description: Enables experimental receive telemetry for Kafka instrumentation.
+    type: boolean
+    default: false