test(kafka): Add an end-to-end test of the raw Kafka client's tracing

SentryKafkaEndToEndTest publishes a message through a wrapped
KafkaProducer to an in-JVM KRaft broker, consumes it with
SentryKafkaConsumerTracing, and checks what reaches MockSentryServer:
the queue.publish span in the producing transaction, and a queue.process
transaction with the consumer origin and the Kafka system that continues
the producer's trace. It ports the Kafka part of the console sample's
system test, which needed a Redpanda container.

The broker comes from spring-kafka-test 3, which needs Java 17, so the
test classpaths resolve Java 17 variants. The tests run on it anyway.

With its core, profiling and Kafka parts ported, the console sample has
no system test left, so it comes off the system-test runner and the
backend workflow, whose base row moves to the console OTLP sample.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Author: runningcode

.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-spring-boot-4-otlp"
             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.core)
   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 {

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,22 +1,12 @@
-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") }
 
-tasks.withType<KotlinCompile>().configureEach {
-  kotlin {
-    compilerOptions.freeCompilerArgs = listOf("-Xjsr305=strict")
-  }
-}
-
 dependencies {
   implementation(projects.sentry)
   implementation(projects.sentryAsyncProfiler)
@@ -25,13 +15,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)
@@ -49,22 +32,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.*") }
-}

sentry-samples/sentry-samples-console/src/test/kotlin/io/sentry/DummyTest.kt

@@ -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"),
         ]