test(kafka): [Sample Tests 34] Add an end-to-end test of the raw Kafka client's tracing

.github/workflows/system-tests-backend.yml

@@ -24,13 +24,10 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        sample: [ "sentry-samples-console" ]
+        sample: [ "sentry-samples-console-otlp" ]
         agent: [ "false" ]
         agent-auto-init: [ "true" ]
         include:
-          - sample: "sentry-samples-console-otlp"
-            agent: "false"
-            agent-auto-init: "true"
           - sample: "sentry-samples-logback"
             agent: "false"
             agent-auto-init: "true"

sentry-kafka/build.gradle.kts

@@ -28,10 +28,13 @@ dependencies {
   // tests
   testImplementation(projects.sentryTestSupport)
   testImplementation(kotlin(Config.kotlinStdLib))
+  testImplementation(libs.google.truth)
   testImplementation(libs.kotlin.test.junit)
   testImplementation(libs.mockito.kotlin)
   testImplementation(libs.mockito.inline)
   testImplementation(libs.kafka.clients)
+  // For an in-JVM broker. It brings no Spring application into the tests.
+  testImplementation(libs.spring.kafka3.test)
 }
 
 tasks.withType<JavaCompile>().configureEach {
@@ -60,3 +63,10 @@ tasks.jar {
     )
   }
 }
+
+// The tests' in-JVM Kafka broker needs Java 17, which the tests run on.
+listOf("testCompileClasspath", "testRuntimeClasspath").forEach {
+  configurations.named(it) {
+    attributes { attribute(TargetJvmVersion.TARGET_JVM_VERSION_ATTRIBUTE, 17) }
+  }
+}

sentry-kafka/src/test/kotlin/io/sentry/kafka/SentryKafkaEndToEndTest.kt

@@ -0,0 +1,103 @@
+package io.sentry.kafka
+
+import com.google.common.truth.Truth.assertThat
+import io.sentry.Sentry
+import io.sentry.test.MockSentryServer
+import io.sentry.test.initForTest
+import io.sentry.test.transaction
+import java.time.Duration
+import java.util.Properties
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import org.apache.kafka.clients.consumer.ConsumerConfig
+import org.apache.kafka.clients.consumer.ConsumerRecord
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.clients.producer.KafkaProducer
+import org.apache.kafka.clients.producer.ProducerConfig
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.apache.kafka.common.serialization.StringDeserializer
+import org.apache.kafka.common.serialization.StringSerializer
+import org.springframework.kafka.test.EmbeddedKafkaKraftBroker
+
+/**
+ * Sends a message through the plain Kafka client to an in-JVM KRaft broker, so no Docker is needed,
+ * and checks that both sides reach Sentry over the real HTTP transport.
+ */
+class SentryKafkaEndToEndTest {
+  private val server = MockSentryServer()
+  private val kafka = EmbeddedKafkaKraftBroker(1, 1, TOPIC).apply { afterPropertiesSet() }
+
+  @AfterTest
+  fun teardown() {
+    Sentry.close()
+    kafka.destroy()
+    server.close()
+  }
+
+  @Test
+  fun `a message is traced where it's published and where it's processed`() {
+    initForTest {
+      it.dsn = server.dsn
+      it.tracesSampleRate = 1.0
+      it.isEnableQueueTracing = true
+    }
+
+    val transaction = Sentry.startTransaction("kafka-demo", "demo")
+    transaction.makeCurrent().use {
+      SentryKafkaProducer.wrap(KafkaProducer<String, String>(producerProperties())).use { producer
+        ->
+        producer.send(ProducerRecord(TOPIC, "sentry-kafka message")).get()
+      }
+    }
+    transaction.finish()
+
+    KafkaConsumer<String, String>(consumerProperties()).use { consumer ->
+      consumer.subscribe(listOf(TOPIC))
+      val record = pollOne(consumer)
+      SentryKafkaConsumerTracing.withTracing(record, Runnable {})
+    }
+
+    val producer =
+      server.awaitEnvelope { it.transaction()?.transaction == "kafka-demo" }.transaction()!!
+    assertThat(producer.spans.map { it.op }).contains("queue.publish")
+
+    val consumer =
+      server
+        .awaitEnvelope { it.transaction()?.contexts?.trace?.operation == "queue.process" }
+        .transaction()!!
+    val trace = consumer.contexts.trace!!
+    assertThat(trace.origin).isEqualTo(SentryKafkaConsumerTracing.TRACE_ORIGIN)
+    assertThat(trace.data).containsEntry("messaging.system", "kafka")
+    // The producer's span puts the trace in the message's headers, and the consumer continues it.
+    assertThat(trace.traceId).isEqualTo(producer.contexts.trace!!.traceId)
+  }
+
+  private fun pollOne(consumer: KafkaConsumer<String, String>): ConsumerRecord<String, String> {
+    repeat(20) {
+      consumer.poll(Duration.ofMillis(500)).firstOrNull()?.let {
+        return it
+      }
+    }
+    error("No message arrived")
+  }
+
+  private fun producerProperties(): Properties =
+    Properties().apply {
+      put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.brokersAsString)
+      put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
+      put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer::class.java.name)
+    }
+
+  private fun consumerProperties(): Properties =
+    Properties().apply {
+      put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.brokersAsString)
+      put(ConsumerConfig.GROUP_ID_CONFIG, "sentry-kafka-end-to-end-test")
+      put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
+      put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
+      put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer::class.java.name)
+    }
+
+  private companion object {
+    const val TOPIC = "sentry-topic"
+  }
+}

sentry-samples/sentry-samples-console/build.gradle.kts

@@ -1,12 +1,8 @@
-import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
-
 plugins {
   java
   application
-  alias(libs.plugins.kotlin.jvm)
   alias(libs.plugins.gradle.versions)
   alias(libs.plugins.shadow)
-  id("io.sentry.systemtest")
 }
 
 application { mainClass.set("io.sentry.samples.console.Main") }
@@ -20,17 +16,6 @@ configure<JavaPluginExtension> {
   targetCompatibility = JavaVersion.VERSION_17
 }
 
-tasks.withType<KotlinCompile>().configureEach {
-  compilerOptions.jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_17
-}
-
-tasks.withType<KotlinCompile>().configureEach {
-  kotlin {
-    compilerOptions.freeCompilerArgs = listOf("-Xjsr305=strict")
-    compilerOptions.jvmTarget = org.jetbrains.kotlin.gradle.dsl.JvmTarget.JVM_17
-  }
-}
-
 dependencies {
   implementation(projects.sentry)
   implementation(projects.sentryAsyncProfiler)
@@ -39,13 +24,6 @@ dependencies {
   implementation(libs.jcache)
   implementation(libs.caffeine.jcache)
   implementation(libs.kafka.clients)
-
-  testImplementation(kotlin(Config.kotlinStdLib))
-  testImplementation(projects.sentry)
-  testImplementation(projects.sentrySystemTestSupport)
-  testImplementation(libs.kotlin.test.junit)
-  testImplementation(libs.slf4j.api)
-  testImplementation(libs.slf4j.jdk14)
 }
 
 // Configure the Shadow JAR (executable JAR with all dependencies)
@@ -63,22 +41,3 @@ tasks.jar {
 
 // Fix the startScripts task dependency
 tasks.startScripts { dependsOn(tasks.shadowJar) }
-
-tasks.register<Test>("systemTest").configure {
-  group = "verification"
-  description = "Runs the System tests"
-
-  val test = project.extensions.getByType<SourceSetContainer>()["test"]
-  testClassesDirs = test.output.classesDirs
-  classpath = test.runtimeClasspath
-
-  maxParallelForks = 1
-
-  filter { includeTestsMatching("io.sentry.systemtest*") }
-}
-
-tasks.named("test").configure {
-  require(this is Test)
-
-  filter { excludeTestsMatching("io.sentry.systemtest.*") }
-}

test/system-test-runner.py

@@ -68,9 +68,7 @@
 
 KAFKA_CONTAINER_NAME = "sentry-java-system-test-kafka"
 KAFKA_BOOTSTRAP_SERVERS = "localhost:9092"
-KAFKA_BROKER_REQUIRED_MODULES = {
-    "sentry-samples-console",
-}
+KAFKA_BROKER_REQUIRED_MODULES = set()
 KAFKA_PROFILE_REQUIRED_MODULES = set()
 
 class ServerType(Enum):
@@ -828,7 +826,6 @@ def get_available_modules(self) -> List[ModuleConfig]:
         """Get list of all available test modules."""
         return [
             ModuleConfig("sentry-samples-spring-boot-4-otlp", "false", "true", "false"),
-            ModuleConfig("sentry-samples-console", "false", "true", "false"),
             ModuleConfig("sentry-samples-console-otlp", "false", "true", "false"),
             ModuleConfig("sentry-samples-console-opentelemetry-noagent", "false", "true", "false"),
             ModuleConfig("sentry-samples-logback", "false", "true", "false"),