support otel.instrumentation.kafka.experimental-span-attributes in spring starter

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/OpenTelemetryAutoConfiguration.java

@@ -19,10 +19,14 @@
 import io.opentelemetry.sdk.autoconfigure.internal.AutoConfigureUtil;
 import io.opentelemetry.sdk.autoconfigure.internal.ComponentLoader;
 import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper;
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
 import io.opentelemetry.sdk.autoconfigure.spi.ResourceProvider;
+import io.opentelemetry.sdk.autoconfigure.spi.internal.DefaultConfigProperties;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.ConfigurationPropertiesBinding;
@@ -80,31 +84,37 @@ public ResourceProvider otelDistroVersionResourceProvider() {
     }
 
     @Bean
-    // If you change the bean name, also change it in the OpenTelemetryJdbcDriverAutoConfiguration
-    // class
-    public OpenTelemetry openTelemetry(
+    public AutoConfiguredOpenTelemetrySdk autoConfiguredOpenTelemetrySdk(
         Environment env,
         OtlpExporterProperties otlpExporterProperties,
         OtelResourceProperties resourceProperties,
         PropagationProperties propagationProperties,
         OpenTelemetrySdkComponentLoader componentLoader) {
 
-      OpenTelemetry openTelemetry =
-          AutoConfigureUtil.setComponentLoader(
-                  AutoConfigureUtil.setConfigPropertiesCustomizer(
-                      AutoConfiguredOpenTelemetrySdk.builder(),
-                      c ->
-                          SpringConfigProperties.create(
-                              env,
-                              otlpExporterProperties,
-                              resourceProperties,
-                              propagationProperties,
-                              c)),
-                  componentLoader)
-              .build()
-              .getOpenTelemetrySdk();
-
-      return openTelemetry;
+      return AutoConfigureUtil.setComponentLoader(
+              AutoConfigureUtil.setConfigPropertiesCustomizer(
+                  AutoConfiguredOpenTelemetrySdk.builder(),
+                  c ->
+                      SpringConfigProperties.create(
+                          env,
+                          otlpExporterProperties,
+                          resourceProperties,
+                          propagationProperties,
+                          c)),
+              componentLoader)
+          .build();
+    }
+
+    @Bean
+    public OpenTelemetry openTelemetry(
+        AutoConfiguredOpenTelemetrySdk autoConfiguredOpenTelemetrySdk) {
+      return autoConfiguredOpenTelemetrySdk.getOpenTelemetrySdk();
+    }
+
+    @Bean
+    public ConfigProperties otelProperties(
+        AutoConfiguredOpenTelemetrySdk autoConfiguredOpenTelemetrySdk) {
+      return AutoConfigureUtil.getConfig(autoConfiguredOpenTelemetrySdk);
     }
   }
 
@@ -117,6 +127,21 @@ public static class DisabledOpenTelemetrySdkConfig {
     public OpenTelemetry openTelemetry() {
       return OpenTelemetry.noop();
     }
+
+    @Bean
+    public ConfigProperties otelProperties() {
+      return DefaultConfigProperties.createFromMap(Collections.emptyMap());
+    }
+  }
+
+  @Configuration
+  @ConditionalOnBean(OpenTelemetry.class)
+  @ConditionalOnMissingBean({ConfigProperties.class})
+  public static class FallbackConfigProperties {
+    @Bean
+    public ConfigProperties otelProperties() {
+      return DefaultConfigProperties.create(Collections.emptyMap());
+    }
   }
 
   /**

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/kafka/ConcurrentKafkaListenerContainerFactoryPostProcessor.java

@@ -7,17 +7,21 @@
 
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.instrumentation.spring.kafka.v2_7.SpringKafkaTelemetry;
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 
 class ConcurrentKafkaListenerContainerFactoryPostProcessor implements BeanPostProcessor {
 
   private final ObjectProvider<OpenTelemetry> openTelemetryProvider;
+  private final ObjectProvider<ConfigProperties> configPropertiesProvider;
 
   ConcurrentKafkaListenerContainerFactoryPostProcessor(
-      ObjectProvider<OpenTelemetry> openTelemetryProvider) {
+      ObjectProvider<OpenTelemetry> openTelemetryProvider,
+      ObjectProvider<ConfigProperties> configPropertiesProvider) {
     this.openTelemetryProvider = openTelemetryProvider;
+    this.configPropertiesProvider = configPropertiesProvider;
   }
 
   @Override
@@ -29,7 +33,12 @@ public Object postProcessAfterInitialization(Object bean, String beanName) {
     ConcurrentKafkaListenerContainerFactory<?, ?> listenerContainerFactory =
         (ConcurrentKafkaListenerContainerFactory<?, ?>) bean;
     SpringKafkaTelemetry springKafkaTelemetry =
-        SpringKafkaTelemetry.create(openTelemetryProvider.getObject());
+        SpringKafkaTelemetry.builder(openTelemetryProvider.getObject())
+            .setCaptureExperimentalSpanAttributes(
+                configPropertiesProvider
+                    .getObject()
+                    .getBoolean("otel.instrumentation.kafka.experimental-span-attributes", false))
+            .build();
     listenerContainerFactory.setBatchInterceptor(springKafkaTelemetry.createBatchInterceptor());
     listenerContainerFactory.setRecordInterceptor(springKafkaTelemetry.createRecordInterceptor());
 

instrumentation/spring/spring-boot-autoconfigure/src/main/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/kafka/KafkaInstrumentationAutoConfiguration.java

@@ -8,6 +8,7 @@
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.instrumentation.kafkaclients.v2_6.KafkaTelemetry;
 import io.opentelemetry.instrumentation.spring.autoconfigure.internal.ConditionalOnEnabledInstrumentation;
+import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties;
 import org.springframework.beans.factory.ObjectProvider;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.kafka.DefaultKafkaProducerFactoryCustomizer;
@@ -32,7 +33,9 @@ DefaultKafkaProducerFactoryCustomizer otelKafkaProducerFactoryCustomizer(
   @Bean
   static ConcurrentKafkaListenerContainerFactoryPostProcessor
       otelKafkaListenerContainerFactoryBeanPostProcessor(
-          ObjectProvider<OpenTelemetry> openTelemetryProvider) {
-    return new ConcurrentKafkaListenerContainerFactoryPostProcessor(openTelemetryProvider);
+          ObjectProvider<OpenTelemetry> openTelemetryProvider,
+          ObjectProvider<ConfigProperties> configPropertiesProvider) {
+    return new ConcurrentKafkaListenerContainerFactoryPostProcessor(
+        openTelemetryProvider, configPropertiesProvider);
   }
 }

instrumentation/spring/spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json

@@ -285,6 +285,12 @@
       "description": "Enable the Kafka instrumentation.",
       "defaultValue": true
     },
+    {
+      "name": "otel.instrumentation.kafka.experimental-span-attributes",
+      "type": "java.lang.Boolean",
+      "description": "Enable the capture of experimental Kafka span attributes.",
+      "defaultValue": false
+    },
     {
       "name": "otel.instrumentation.log4j-appender.enabled",
       "type": "java.lang.Boolean",

instrumentation/spring/spring-boot-autoconfigure/src/test/java/io/opentelemetry/instrumentation/spring/autoconfigure/OpenTelemetryAutoConfigurationTest.java

@@ -48,7 +48,8 @@ void customOpenTelemetry() {
             context ->
                 assertThat(context)
                     .hasBean("customOpenTelemetry")
-                    .doesNotHaveBean("openTelemetry"));
+                    .doesNotHaveBean("openTelemetry")
+                    .hasBean("otelProperties"));
   }
 
   @Test
@@ -57,7 +58,7 @@ void customOpenTelemetry() {
   void initializeProvidersAndOpenTelemetry() {
     this.contextRunner
         .withConfiguration(AutoConfigurations.of(OpenTelemetryAutoConfiguration.class))
-        .run(context -> assertThat(context).hasBean("openTelemetry"));
+        .run(context -> assertThat(context).hasBean("openTelemetry").hasBean("otelProperties"));
   }
 
   @Test

smoke-tests-otel-starter/spring-boot-common/src/main/java/io/opentelemetry/spring/smoketest/AbstractJvmKafkaSpringStarterSmokeTest.java

@@ -55,6 +55,7 @@ void setUpContext() {
                     KafkaInstrumentationAutoConfiguration.class,
                     KafkaConfig.class))
             .withPropertyValues(
+                "otel.instrumentation.kafka.experimental-span-attributes=true",
                 "spring.kafka.bootstrap-servers=" + kafka.getBootstrapServers(),
                 "spring.kafka.consumer.auto-offset-reset=earliest",
                 "spring.kafka.consumer.linger-ms=10",

smoke-tests-otel-starter/spring-boot-common/src/main/java/io/opentelemetry/spring/smoketest/AbstractKafkaSpringStarterSmokeTest.java

@@ -9,6 +9,7 @@
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 
 import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes;
 import org.apache.kafka.clients.admin.NewTopic;
@@ -97,6 +98,9 @@ void shouldInstrumentProducerAndConsumer() {
                             equalTo(
                                 MessagingIncubatingAttributes.MESSAGING_KAFKA_CONSUMER_GROUP,
                                 "testListener"),
+                            satisfies(
+                                AttributeKey.longKey("kafka.record.queue_time_ms"),
+                                AbstractLongAssert::isNotNegative),
                             satisfies(
                                 MessagingIncubatingAttributes.MESSAGING_CLIENT_ID,
                                 stringAssert -> stringAssert.startsWith("consumer"))),

smoke-tests-otel-starter/spring-boot-common/src/main/resources/application.yaml

@@ -1,5 +1,7 @@
 otel:
   instrumentation:
+    kafka:
+      experimental-span-attributes: true
     logback-appender:
       experimental:
         capture-code-attributes: true