don't init kafka in other tests

Author: zeitlinger

smoke-tests-otel-starter/spring-boot-3/build.gradle.kts

@@ -16,8 +16,7 @@ dependencies {
   runtimeOnly("com.h2database:h2")
   implementation("org.apache.commons:commons-dbcp2")
   implementation(project(":instrumentation:jdbc:library"))
-  implementation("org.springframework.kafka:spring-kafka") // not tested here, just make sure there are no warnings when it's included
-  implementation("org.springframework.kafka:spring-kafka:2.9.0")
+  implementation("org.springframework.kafka:spring-kafka")
   implementation("io.opentelemetry:opentelemetry-extension-trace-propagators")
   implementation(project(":instrumentation:spring:starters:spring-boot-starter"))
   implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES))

smoke-tests-otel-starter/spring-boot-3/src/main/java/io/opentelemetry/spring/smoketest/KafkaConfig.java

@@ -8,12 +8,9 @@
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.metrics.LongHistogram;
 import io.opentelemetry.api.metrics.Meter;
-import io.opentelemetry.api.trace.SpanKind;
 import java.util.Optional;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.springframework.boot.web.client.RestTemplateBuilder;
 import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
-import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.client.RestClient;
@@ -31,14 +28,12 @@ public class OtelSpringStarterSmokeTestController {
   private final LongHistogram histogram;
   private final Optional<RestTemplate> restTemplate;
   private final Optional<RestClient> restClient;
-  private final OpenTelemetry openTelemetry;
 
   public OtelSpringStarterSmokeTestController(
       OpenTelemetry openTelemetry,
       RestClient.Builder restClientBuilder,
       RestTemplateBuilder restTemplateBuilder,
       Optional<ServletWebServerApplicationContext> server) {
-    this.openTelemetry = openTelemetry;
     Meter meter = openTelemetry.getMeter(METER_SCOPE_NAME);
     histogram = meter.histogramBuilder(TEST_HISTOGRAM).ofLongs().build();
     Optional<String> rootUri = server.map(s -> "http://localhost:" + s.getWebServer().getPort());
@@ -65,10 +60,4 @@ public String restTemplate() {
         .map(t -> t.getForObject(PING, String.class))
         .orElseThrow(() -> new IllegalStateException("RestTemplate not available"));
   }
-
-  @KafkaListener(id = "testListener", topics = "testTopic")
-  public void listener(ConsumerRecord<String, String> record) {
-    openTelemetry.getTracer("consumer", "1.0").spanBuilder("consumer").setSpanKind(SpanKind.CONSUMER)
-        .startSpan().end();
-  }
 }

smoke-tests-otel-starter/spring-boot-3/src/main/java/io/opentelemetry/spring/smoketest/OtelSpringStarterSmokeTestController.java

@@ -8,43 +8,55 @@
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
 
+import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes;
 import io.opentelemetry.spring.smoketest.AbstractSpringStarterSmokeTest;
 import io.opentelemetry.spring.smoketest.OtelSpringStarterSmokeTestApplication;
 import io.opentelemetry.spring.smoketest.SpringSmokeOtelConfiguration;
+import java.time.Duration;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.assertj.core.api.AbstractLongAssert;
 import org.assertj.core.api.AbstractStringAssert;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.kafka.config.TopicBuilder;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.testcontainers.containers.KafkaContainer;
 import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.junit.jupiter.Container;
 import org.testcontainers.junit.jupiter.Testcontainers;
 import org.testcontainers.utility.DockerImageName;
-import java.time.Duration;
 
 @Testcontainers
 @SpringBootTest(
-    classes = {OtelSpringStarterSmokeTestApplication.class, SpringSmokeOtelConfiguration.class},
-    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, properties = {
+    classes = {
+      OtelSpringStarterSmokeTestApplication.class,
+      SpringSmokeOtelConfiguration.class,
+      KafkaSpringStarterSmokeTest.KafkaConfig.class
+    },
+    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT,
+    properties = {
       "spring.kafka.consumer.auto-offset-reset=earliest",
       "spring.kafka.consumer.linger-ms=10",
       "spring.kafka.admin.operation-timeout=5m",
       "spring.kafka.listener.idle-between-polls=1000",
       "spring.kafka.producer.transaction-id-prefix=test-"
-
-})
+    })
 public class KafkaSpringStarterSmokeTest extends AbstractSpringStarterSmokeTest {
 
   @Container @ServiceConnection
-  static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.10"))
-      .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m")
-      .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
-      .withStartupTimeout(Duration.ofMinutes(1));
+  static KafkaContainer kafka =
+      new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.10"))
+          .withEnv("KAFKA_HEAP_OPTS", "-Xmx256m")
+          .waitingFor(Wait.forLogMessage(".*started \\(kafka.server.KafkaServer\\).*", 1))
+          .withStartupTimeout(Duration.ofMinutes(1));
 
   @Autowired private KafkaTemplate<String, String> kafkaTemplate;
 
@@ -119,4 +131,25 @@ void shouldInstrumentProducerAndConsumer() {
                                 stringAssert -> stringAssert.startsWith("consumer"))),
                 span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
   }
+
+  @Configuration
+  public static class KafkaConfig {
+
+    @Autowired OpenTelemetry openTelemetry;
+
+    @Bean
+    public NewTopic testTopic() {
+      return TopicBuilder.name("testTopic").partitions(1).replicas(1).build();
+    }
+
+    @KafkaListener(id = "testListener", topics = "testTopic")
+    public void listener(ConsumerRecord<String, String> record) {
+      openTelemetry
+          .getTracer("consumer", "1.0")
+          .spanBuilder("consumer")
+          .setSpanKind(SpanKind.CONSUMER)
+          .startSpan()
+          .end();
+    }
+  }
 }