test(samples): Add Kafka queue system tests for Spring Boot 3

Add KafkaQueueSystemTest with e2e tests for:
- Producer endpoint creates queue.publish span
- Consumer creates queue.process transaction
- Distributed tracing (producer and consumer share same trace)
- Messaging attributes on publish span and process transaction

Also add produceKafkaMessage to RestTestClient and enable
sentry.enable-queue-tracing in the kafka profile properties.

Requires a running Kafka broker at localhost:9092 and the sample app
started with --spring.profiles.active=kafka.

Co-Authored-By: Claude <noreply@anthropic.com>

Author: adinauer

sentry-samples/sentry-samples-spring-boot-jakarta/src/main/resources/application-kafka.properties

@@ -1,4 +1,6 @@
 # Kafka — activate with: --spring.profiles.active=kafka
+sentry.enable-queue-tracing=true
+
 spring.autoconfigure.exclude=
 spring.kafka.bootstrap-servers=localhost:9092
 spring.kafka.consumer.group-id=sentry-sample-group

sentry-samples/sentry-samples-spring-boot-jakarta/src/test/kotlin/io/sentry/systemtest/KafkaQueueSystemTest.kt

@@ -0,0 +1,111 @@
+package io.sentry.systemtest
+
+import io.sentry.systemtest.util.TestHelper
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import org.junit.Before
+
+/**
+ * System tests for Kafka queue instrumentation.
+ *
+ * Requires:
+ * - The sample app running with `--spring.profiles.active=kafka`
+ * - A Kafka broker at localhost:9092
+ * - The mock Sentry server at localhost:8000
+ */
+class KafkaQueueSystemTest {
+  lateinit var testHelper: TestHelper
+
+  @Before
+  fun setup() {
+    testHelper = TestHelper("http://localhost:8080")
+    testHelper.reset()
+  }
+
+  @Test
+  fun `producer endpoint creates queue publish span`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("test-message")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      testHelper.doesTransactionContainSpanWithOp(transaction, "queue.publish")
+    }
+  }
+
+  @Test
+  fun `consumer creates queue process transaction`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("test-consumer-message")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    // The consumer runs asynchronously, so wait for the queue.process transaction
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      testHelper.doesTransactionHaveOp(transaction, "queue.process")
+    }
+  }
+
+  @Test
+  fun `producer and consumer share same trace`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("trace-test-message")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    // Capture the trace ID from the producer transaction
+    var producerTraceId: String? = null
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      if (testHelper.doesTransactionContainSpanWithOp(transaction, "queue.publish")) {
+        producerTraceId = transaction.contexts.trace?.traceId?.toString()
+        true
+      } else {
+        false
+      }
+    }
+
+    // Verify the consumer transaction has the same trace ID
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      if (testHelper.doesTransactionHaveOp(transaction, "queue.process")) {
+        val consumerTraceId = transaction.contexts.trace?.traceId?.toString()
+        producerTraceId != null && consumerTraceId == producerTraceId
+      } else {
+        false
+      }
+    }
+  }
+
+  @Test
+  fun `queue publish span has messaging attributes`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("attrs-test")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      val span = transaction.spans.firstOrNull { it.op == "queue.publish" }
+      if (span == null) return@ensureTransactionReceived false
+
+      val data = span.data ?: return@ensureTransactionReceived false
+      data["messaging.system"] == "kafka" && data["messaging.destination.name"] == "sentry-topic"
+    }
+  }
+
+  @Test
+  fun `queue process transaction has messaging attributes`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("process-attrs-test")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      if (!testHelper.doesTransactionHaveOp(transaction, "queue.process")) {
+        return@ensureTransactionReceived false
+      }
+
+      val data = transaction.contexts.trace?.data ?: return@ensureTransactionReceived false
+      data["messaging.system"] == "kafka" && data["messaging.destination.name"] == "sentry-topic"
+    }
+  }
+}

sentry-system-test-support/api/sentry-system-test-support.api

@@ -560,6 +560,8 @@ public final class io/sentry/systemtest/util/RestTestClient : io/sentry/systemte
 	public final fun getTodo (J)Lio/sentry/systemtest/Todo;
 	public final fun getTodoRestClient (J)Lio/sentry/systemtest/Todo;
 	public final fun getTodoWebclient (J)Lio/sentry/systemtest/Todo;
+	public final fun produceKafkaMessage (Ljava/lang/String;)Ljava/lang/String;
+	public static synthetic fun produceKafkaMessage$default (Lio/sentry/systemtest/util/RestTestClient;Ljava/lang/String;ILjava/lang/Object;)Ljava/lang/String;
 	public final fun saveCachedTodo (Lio/sentry/systemtest/Todo;)Lio/sentry/systemtest/Todo;
 }
 

sentry-system-test-support/src/main/kotlin/io/sentry/systemtest/util/RestTestClient.kt

@@ -81,6 +81,12 @@ class RestTestClient(private val backendBaseUrl: String) : LoggingInsecureRestCl
     return response?.body?.string()
   }
 
+  fun produceKafkaMessage(message: String = "hello from sentry!"): String? {
+    val request = Request.Builder().url("$backendBaseUrl/kafka/produce?message=$message")
+
+    return callTyped(request, true)
+  }
+
   fun getCountMetric(): String? {
     val request = Request.Builder().url("$backendBaseUrl/metric/count")