add kafka Spring starter smoke test (#11262)

Co-authored-by: Jean Bisutti <jean.bisutti@gmail.com>

.github/graal-native-docker-compose.yaml

@@ -0,0 +1,22 @@
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:6.2.10
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181
+      ZOOKEEPER_TICK_TIME: 2000
+    ports:
+      - "22181:2181"
+
+  kafka:
+    image: confluentinc/cp-kafka:6.2.10
+    depends_on:
+      - zookeeper
+    ports:
+      - "29092:29092"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

.github/workflows/reusable-native-tests.yml

@@ -19,9 +19,16 @@ jobs:
           java-version: "21"
           components: "native-image"
       - name: Running test
+        env:
+          DOCKER_COMPOSE_TEST: "true"
         run: |
           echo "GRAALVM_HOME: $GRAALVM_HOME"
           echo "JAVA_HOME: $JAVA_HOME"
           java --version
           native-image --version
+          # Testcontainers does not work in some cases with GraalVM native images, 
+          # therefore we're starting a Kafka container manually for the tests
+          docker compose -f .github/graal-native-docker-compose.yaml up -d
+          # don't wait for startup - gradle compile takes long enough
           ./gradlew nativeTest
+          docker compose -f .github/graal-native-docker-compose.yaml down # is this needed?

instrumentation/spring/spring-boot-autoconfigure/build.gradle.kts

@@ -67,7 +67,6 @@ dependencies {
   testLibrary("org.springframework.boot:spring-boot-starter-test:$springBootVersion") {
     exclude("org.junit.vintage", "junit-vintage-engine")
   }
-  testImplementation("org.testcontainers:kafka")
   testImplementation("javax.servlet:javax.servlet-api:3.1.0")
   testImplementation("jakarta.servlet:jakarta.servlet-api:5.0.0")
 
@@ -151,10 +150,6 @@ tasks {
     options.compilerArgs.add("-parameters")
   }
 
-  test {
-    usesService(gradle.sharedServices.registrations["testcontainersBuildService"].service)
-  }
-
   withType<Test>().configureEach {
     systemProperty("testLatestDeps", latestDepTest)
 

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

@@ -10,10 +10,13 @@ dependencies {
   implementation("org.springframework.boot:spring-boot-starter-data-jdbc")
   runtimeOnly("com.h2database:h2")
   implementation("org.apache.commons:commons-dbcp2")
-  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")
   implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES))
 
   implementation(project(":smoke-tests-otel-starter:spring-boot-common"))
+
+  testImplementation("org.testcontainers:junit-jupiter")
+  testImplementation("org.testcontainers:kafka")
   testImplementation("org.springframework.boot:spring-boot-starter-test")
 }
 

smoke-tests-otel-starter/spring-boot-2/src/test/java/io/opentelemetry/spring/smoketest/KafkaSpringStarterSmokeTest.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.spring.smoketest;
+
+import org.junit.jupiter.api.condition.DisabledInNativeImage;
+
+@DisabledInNativeImage // See GraalVmNativeKafkaSpringStarterSmokeTest for the GraalVM native test
+public class KafkaSpringStarterSmokeTest extends AbstractJvmKafkaSpringStarterSmokeTest {}

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

@@ -15,10 +15,13 @@ dependencies {
   implementation("org.springframework.boot:spring-boot-starter-data-jdbc")
   runtimeOnly("com.h2database:h2")
   implementation("org.apache.commons:commons-dbcp2")
-  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")
   implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES))
 
   implementation(project(":smoke-tests-otel-starter:spring-boot-common"))
+
+  testImplementation("org.testcontainers:junit-jupiter")
+  testImplementation("org.testcontainers:kafka")
   testImplementation("org.springframework.boot:spring-boot-starter-test")
 }
 

smoke-tests-otel-starter/spring-boot-3/src/test/java/io/opentelemetry/spring/smoketest/GraalVmNativeKafkaSpringStarterSmokeTest.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.spring.smoketest;
+
+import org.junit.jupiter.api.condition.EnabledInNativeImage;
+import org.springframework.boot.test.context.SpringBootTest;
+
+/**
+ * GraalVM native image doesn't support Testcontainers in our case, so the docker container is
+ * started manually before running the tests.
+ *
+ * <p>In other cases, it does work, e.g. <a
+ * href="https://info.michael-simons.eu/2023/10/25/run-your-integration-tests-against-testcontainers-with-graalvm-native-image/">here</a>,
+ * it's not yet clear why it doesn't work in our case.
+ *
+ * <p>In CI, this is done in reusable-native-tests.yml. If you want to run the tests locally, you
+ * need to start the container manually: see .github/workflows/reusable-native-tests.yml for the
+ * command.
+ */
+@SpringBootTest(
+    classes = {
+      OtelSpringStarterSmokeTestApplication.class,
+      SpringSmokeOtelConfiguration.class,
+      AbstractKafkaSpringStarterSmokeTest.KafkaConfig.class
+    },
+    webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
+@EnabledInNativeImage // see JvmMongodbSpringStarterSmokeTest for the JVM test
+@RequiresDockerComposeEnvVariable
+public class GraalVmNativeKafkaSpringStarterSmokeTest extends AbstractKafkaSpringStarterSmokeTest {}

smoke-tests-otel-starter/spring-boot-3/src/test/java/io/opentelemetry/spring/smoketest/KafkaSpringStarterSmokeTest.java

@@ -0,0 +1,11 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.spring.smoketest;
+
+import org.junit.jupiter.api.condition.DisabledInNativeImage;
+
+@DisabledInNativeImage // See GraalVmNativeKafkaSpringStarterSmokeTest for the GraalVM native test
+public class KafkaSpringStarterSmokeTest extends AbstractJvmKafkaSpringStarterSmokeTest {}

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

@@ -14,6 +14,9 @@ dependencies {
   compileOnly("org.springframework.boot:spring-boot-starter-test")
   compileOnly("org.springframework.boot:spring-boot-starter-data-jdbc")
   compileOnly("org.apache.commons:commons-dbcp2")
+  compileOnly("org.springframework.kafka:spring-kafka")
+  compileOnly("org.testcontainers:junit-jupiter")
+  compileOnly("org.testcontainers:kafka")
 
   api(project(":smoke-tests-otel-starter:spring-smoke-testing"))
 

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

@@ -0,0 +1,76 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.spring.smoketest;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.instrumentation.spring.autoconfigure.OpenTelemetryAutoConfiguration;
+import io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.kafka.KafkaInstrumentationAutoConfiguration;
+import java.time.Duration;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.testcontainers.containers.KafkaContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+
+/** Spring has a test container integration, but that doesn't work for Spring Boot 2 */
+public class AbstractJvmKafkaSpringStarterSmokeTest extends AbstractKafkaSpringStarterSmokeTest {
+  static KafkaContainer kafka;
+
+  private ApplicationContextRunner contextRunner;
+
+  @BeforeAll
+  static void setUpKafka() {
+    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));
+    kafka.start();
+  }
+
+  @AfterAll
+  static void tearDownKafka() {
+    kafka.stop();
+  }
+
+  @BeforeEach
+  void setUpContext() {
+    contextRunner =
+        new ApplicationContextRunner()
+            .withAllowBeanDefinitionOverriding(true)
+            .withConfiguration(
+                AutoConfigurations.of(
+                    OpenTelemetryAutoConfiguration.class,
+                    SpringSmokeOtelConfiguration.class,
+                    KafkaAutoConfiguration.class,
+                    KafkaInstrumentationAutoConfiguration.class,
+                    KafkaConfig.class))
+            .withPropertyValues(
+                "spring.kafka.bootstrap-servers=" + kafka.getBootstrapServers(),
+                "spring.kafka.consumer.auto-offset-reset=earliest",
+                "spring.kafka.consumer.linger-ms=10",
+                "spring.kafka.listener.idle-between-polls=1000",
+                "spring.kafka.producer.transaction-id-prefix=test-");
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  @Test
+  void shouldInstrumentProducerAndConsumer() {
+    contextRunner.run(
+        applicationContext -> {
+          testing = new SpringSmokeTestRunner(applicationContext.getBean(OpenTelemetry.class));
+          kafkaTemplate = applicationContext.getBean(KafkaTemplate.class);
+          super.shouldInstrumentProducerAndConsumer();
+        });
+  }
+}

instrumentation/spring/spring-boot-autoconfigure/src/test/java/io/opentelemetry/instrumentation/spring/autoconfigure/instrumentation/kafka/KafkaIntegrationTest.java → smoke-tests-otel-starter/spring-boot-common/src/main/java/io/opentelemetry/spring/smoketest/AbstractKafkaSpringStarterSmokeTest.java

@@ -3,98 +3,47 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package io.opentelemetry.instrumentation.spring.autoconfigure.instrumentation.kafka;
+package io.opentelemetry.spring.smoketest;
 
 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.instrumentation.testing.junit.LibraryInstrumentationExtension;
 import io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes;
-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.AfterAll;
-import org.junit.jupiter.api.BeforeAll;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-import org.springframework.boot.autoconfigure.AutoConfigurations;
-import org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration;
-import org.springframework.boot.test.context.runner.ApplicationContextRunner;
-import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.beans.factory.annotation.Autowired;
 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.utility.DockerImageName;
 
-class KafkaIntegrationTest {
+abstract class AbstractKafkaSpringStarterSmokeTest extends AbstractSpringStarterSmokeTest {
 
-  @RegisterExtension
-  static final LibraryInstrumentationExtension testing = LibraryInstrumentationExtension.create();
-
-  static KafkaContainer kafka;
-
-  private ApplicationContextRunner contextRunner;
-
-  @BeforeAll
-  static void setUpKafka() {
-    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));
-    kafka.start();
-  }
-
-  @AfterAll
-  static void tearDownKafka() {
-    kafka.stop();
-  }
-
-  @BeforeEach
-  void setUpContext() {
-    contextRunner =
-        new ApplicationContextRunner()
-            .withConfiguration(
-                AutoConfigurations.of(
-                    KafkaAutoConfiguration.class,
-                    KafkaInstrumentationAutoConfiguration.class,
-                    TestConfig.class))
-            .withBean("openTelemetry", OpenTelemetry.class, testing::getOpenTelemetry)
-            .withPropertyValues(
-                "spring.kafka.bootstrap-servers=" + kafka.getBootstrapServers(),
-                "spring.kafka.consumer.auto-offset-reset=earliest",
-                "spring.kafka.consumer.linger-ms=10",
-                "spring.kafka.listener.idle-between-polls=1000",
-                "spring.kafka.producer.transaction-id-prefix=test-");
-  }
+  @Autowired protected KafkaTemplate<String, String> kafkaTemplate;
 
   @Test
   void shouldInstrumentProducerAndConsumer() {
-    contextRunner.run(KafkaIntegrationTest::runShouldInstrumentProducerAndConsumer);
-  }
-
-  // In kafka 2 ops.send is deprecated. We are using it to avoid reflection because kafka 3 also has
-  // ops.send, although with different return type.
-  @SuppressWarnings({"unchecked", "deprecation"})
-  private static void runShouldInstrumentProducerAndConsumer(
-      ConfigurableApplicationContext applicationContext) {
-    KafkaTemplate<String, String> kafkaTemplate = applicationContext.getBean(KafkaTemplate.class);
+    testing.clearAllExportedData(); // ignore data from application startup
 
     testing.runWithSpan(
         "producer",
         () -> {
           kafkaTemplate.executeInTransaction(
               ops -> {
-                ops.send("testTopic", "10", "testSpan");
+                // return type is incompatible between Spring Boot 2 and 3
+                try {
+                  ops.getClass()
+                      .getDeclaredMethod("send", String.class, Object.class, Object.class)
+                      .invoke(ops, "testTopic", "10", "testSpan");
+                } catch (Exception e) {
+                  throw new IllegalStateException(e);
+                }
                 return 0;
               });
         });
@@ -128,7 +77,7 @@ class KafkaIntegrationTest {
                     span.hasName("testTopic process")
                         .hasKind(SpanKind.CONSUMER)
                         .hasParent(trace.getSpan(1))
-                        .hasAttributesSatisfyingExactly(
+                        .hasAttributesSatisfying(
                             equalTo(MessagingIncubatingAttributes.MESSAGING_SYSTEM, "kafka"),
                             equalTo(
                                 MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME,
@@ -155,7 +104,9 @@ class KafkaIntegrationTest {
   }
 
   @Configuration
-  static class TestConfig {
+  public static class KafkaConfig {
+
+    @Autowired OpenTelemetry openTelemetry;
 
     @Bean
     public NewTopic testTopic() {
@@ -164,7 +115,12 @@ class KafkaIntegrationTest {
 
     @KafkaListener(id = "testListener", topics = "testTopic")
     public void listener(ConsumerRecord<String, String> record) {
-      testing.runWithSpan("consumer", () -> {});
+      openTelemetry
+          .getTracer("consumer", "1.0")
+          .spanBuilder("consumer")
+          .setSpanKind(SpanKind.CONSUMER)
+          .startSpan()
+          .end();
     }
   }
 }

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

@@ -8,3 +8,12 @@ otel:
   resource:
     attributes:
       attributeFromYaml: true # boolean will be automatically converted to string by spring
+
+spring:
+  kafka:
+    consumer:
+      auto-offset-reset: earliest
+    listener:
+      idle-between-polls: 1000
+    producer:
+      transaction-id-prefix: test-

smoke-tests-otel-starter/spring-smoke-testing/src/main/java/io/opentelemetry/spring/smoketest/AbstractSpringStarterSmokeTest.java

@@ -9,6 +9,7 @@ import static org.assertj.core.api.Assertions.assertThat;
 
 import io.opentelemetry.api.OpenTelemetry;
 import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -22,9 +23,19 @@ public abstract class AbstractSpringStarterSmokeTest {
 
   protected SpringSmokeTestRunner testing;
 
+  @BeforeAll
+  static void beforeAll() {
+    SpringSmokeTestRunner.resetExporters();
+  }
+
   @BeforeEach
   void setUpTesting() {
-    testing = new SpringSmokeTestRunner(openTelemetry);
+    if (openTelemetry != null) {
+      // @Autowired doesn't work in all tests, e.g. AbstractJvmKafkaSpringStarterSmokeTest
+      // those tests have to manage the testing instance,
+      // themselves because they don't use @SpringBootTest
+      testing = new SpringSmokeTestRunner(openTelemetry);
+    }
   }
 
   @AfterEach

smoke-tests-otel-starter/spring-smoke-testing/src/main/java/io/opentelemetry/spring/smoketest/RequiresDockerComposeEnvVariable.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.spring.smoketest;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
+
+@Target({ElementType.TYPE, ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Documented
+@EnabledIfEnvironmentVariable(
+    named = "DOCKER_COMPOSE_TEST",
+    matches = "true",
+    disabledReason =
+        "Testcontainers does not work in some cases with GraalVM native images. "
+            + "A container has to be started manually. "
+            + "So, an environment variable is used to disable the test by default.")
+public @interface RequiresDockerComposeEnvVariable {}

smoke-tests-otel-starter/spring-smoke-testing/src/main/java/io/opentelemetry/spring/smoketest/SpringSmokeTestRunner.java

@@ -22,16 +22,22 @@ import java.util.List;
  */
 public final class SpringSmokeTestRunner extends InstrumentationTestRunner {
 
-  static final InMemorySpanExporter testSpanExporter = InMemorySpanExporter.create();
-  static final InMemoryMetricExporter testMetricExporter =
-      InMemoryMetricExporter.create(AggregationTemporality.DELTA);
-  static final InMemoryLogRecordExporter testLogRecordExporter = InMemoryLogRecordExporter.create();
+  static InMemorySpanExporter testSpanExporter;
+  static InMemoryMetricExporter testMetricExporter;
+  static InMemoryLogRecordExporter testLogRecordExporter;
+
   static OpenTelemetry openTelemetry;
 
   public SpringSmokeTestRunner(OpenTelemetry openTelemetry) {
     super(openTelemetry);
   }
 
+  static void resetExporters() {
+    testSpanExporter = InMemorySpanExporter.create();
+    testMetricExporter = InMemoryMetricExporter.create(AggregationTemporality.DELTA);
+    testLogRecordExporter = InMemoryLogRecordExporter.create();
+  }
+
   @Override
   public void beforeTestClass() {}