feat(spring): [Queue Instrumentation 41] Add Spring Boot 2 Kafka tracing

.cursor/rules/overview_dev.mdc

@@ -66,6 +66,15 @@ Use the `fetch_rules` tool to include these rules when working on specific areas
   - `SentryMetricsEvent`, `SentryMetricsEvents`
   - `SentryOptions.getMetrics()`, `beforeSend` callback
 
+- **`queues`**: Use when working with:
+  - Sentry Queues product data or messaging span conventions
+  - Queue tracing spans/transactions (`queue.publish`, `queue.process`)
+  - `enableQueueTracing` option and `sentry.enable-queue-tracing`
+  - Kafka instrumentation (`sentry-kafka`, `SentryKafkaProducer`, `SentryKafkaConsumerTracing`)
+  - Spring Kafka queue auto-instrumentation and `SentryKafkaRecordInterceptor`
+  - Messaging span data (`messaging.system`, `messaging.destination.name`, receive latency, retry count)
+  - `sentry-task-enqueued-time` header and distributed trace propagation through queues
+
 - **`continuous_profiling_jvm`**: Use when working with:
   - JVM continuous profiling (`sentry-async-profiler` module)
   - `IContinuousProfiler`, `JavaContinuousProfiler`
@@ -118,6 +127,7 @@ Use the `fetch_rules` tool to include these rules when working on specific areas
    - System test/e2e/sample → `e2e_tests`
    - Feature flag/addFeatureFlag/flag evaluation → `feature_flags`
    - Metrics/count/distribution/gauge → `metrics`
+   - Queues/queue tracing/Kafka/Spring Kafka/queue.publish/queue.process/enableQueueTracing/messaging spans → `queues`
    - PR/pull request/stacked PR/stack → `pr`
    - JVM continuous profiling/async-profiler/JFR/ProfileChunk → `continuous_profiling_jvm`
    - Android continuous profiling/AndroidProfiler/frame metrics/method tracing → no dedicated rule yet; inspect the code directly

.cursor/rules/queues.mdc

@@ -0,0 +1,82 @@
+---
+alwaysApply: false
+description: Sentry Queues module and Java SDK queue tracing
+---
+# Sentry Queues and Java SDK Queue Tracing
+
+## Product model
+
+Sentry Queues is built from tracing data. SDKs mark queue work with queue-specific span operations and messaging span data so Sentry can identify producers, consumers, destinations, latency, and failures.
+
+The important concepts are:
+- `queue.publish`: a span for enqueueing/publishing a message to a queue or topic.
+- `queue.process`: a transaction for processing a dequeued message.
+- Messaging span data, especially:
+  - `messaging.system` (for example `kafka`)
+  - `messaging.destination.name` (queue/topic name)
+  - `messaging.message.id`
+  - `messaging.message.retry.count`
+  - `messaging.message.body.size`
+  - `messaging.message.envelope.size`
+  - `messaging.message.receive.latency`
+- Distributed tracing headers (`sentry-trace` and `baggage`) link producer-side work to consumer-side processing.
+- Queue receive latency is the time a message spent waiting between publish/enqueue and processing. For Java Kafka, this comes from the `sentry-task-enqueued-time` header that the producer writes and the consumer reads.
+
+The Queues UI is not backed by a separate Java event type. The Java SDK contributes data through spans/transactions with the expected operations, trace context, statuses, and messaging attributes.
+
+## Java SDK implementation
+
+Queue tracing is opt-in. `SentryOptions.isEnableQueueTracing()` defaults to `false` and can be enabled with `setEnableQueueTracing(true)` or external config key `enable-queue-tracing` (`sentry.enable-queue-tracing` in Spring Boot). Captured queue spans/transactions still depend on tracing being enabled and sampled.
+
+Kafka support lives in `sentry-kafka`:
+- `SentryKafkaProducer.wrap(Producer)` wraps Kafka `Producer.send(...)` calls.
+  - Creates a `queue.publish` child span when there is an active span.
+  - Sets `messaging.system=kafka` and `messaging.destination.name=<topic>`.
+  - Injects `sentry-trace`, `baggage`, and `sentry-task-enqueued-time` headers.
+  - Still injects tracing/enqueued-time headers when queue tracing is enabled but there is no active span, so background producers can link to consumers.
+  - Finishes the span from the Kafka callback with `OK` or `INTERNAL_ERROR`.
+- `SentryKafkaConsumerTracing.withTracing(record, callback)` is the manual raw-Kafka consumer helper.
+  - Forks root scopes for the processing lifecycle and makes them current.
+  - Continues the trace from Kafka headers.
+  - Starts a `queue.process` transaction bound to scope when tracing is enabled.
+  - Sets Kafka messaging data, body size, retry count, and receive latency when available.
+  - Finishes with `OK` or `INTERNAL_ERROR` and never lets instrumentation failures break customer processing.
+
+Spring Kafka support lives in `sentry-spring`, `sentry-spring-jakarta`, and `sentry-spring-7`:
+- `SentryKafkaProducerBeanPostProcessor` installs a producer post-processor on `DefaultKafkaProducerFactory` and wraps created producers with `SentryKafkaProducer.wrap(...)`.
+- `SentryKafkaConsumerBeanPostProcessor` installs `SentryKafkaRecordInterceptor` on listener container factories.
+- `SentryKafkaRecordInterceptor` starts/finishes `queue.process` transactions around listener processing, continues traces from headers, forks scopes for the record lifecycle, and preserves any existing delegate interceptor.
+- Spring Boot auto-configuration registers both post-processors only when Spring Kafka and `sentry-kafka` are present and `sentry.enable-queue-tracing=true`.
+- Spring Boot queue auto-configuration is disabled when Sentry OpenTelemetry integration classes are present to avoid duplicate Kafka instrumentation.
+
+## Trace origins and suppression
+
+Queue instrumentation sets span origins so it can be identified and suppressed with `ignoredSpanOrigins`:
+- Raw Kafka producer: `auto.queue.kafka.producer`
+- Raw Kafka consumer helper: `manual.queue.kafka.consumer`
+- Spring Kafka producer: `auto.queue.spring.kafka.producer`, `auto.queue.spring_jakarta.kafka.producer`, `auto.queue.spring7.kafka.producer`
+- Spring Kafka consumer: `auto.queue.spring.kafka.consumer`, `auto.queue.spring_jakarta.kafka.consumer`, `auto.queue.spring7.kafka.consumer`
+
+## Files to inspect when changing queue tracing
+
+- Core option and conventions:
+  - `sentry/src/main/java/io/sentry/SentryOptions.java`
+  - `sentry/src/main/java/io/sentry/ExternalOptions.java`
+  - `sentry/src/main/java/io/sentry/SpanDataConvention.java`
+- Raw Kafka:
+  - `sentry-kafka/src/main/java/io/sentry/kafka/SentryKafkaProducer.java`
+  - `sentry-kafka/src/main/java/io/sentry/kafka/SentryKafkaConsumerTracing.java`
+  - `sentry-kafka/src/test/kotlin/io/sentry/kafka/*Test.kt`
+- Spring Kafka:
+  - `sentry-spring*/src/main/java/io/sentry/**/kafka/*`
+  - `sentry-spring*/src/test/kotlin/io/sentry/**/kafka/*Test.kt`
+  - `sentry-spring-boot*/src/main/java/io/sentry/**/SentryAutoConfiguration.java`
+  - `sentry-spring-boot*/src/test/kotlin/io/sentry/**/SentryKafkaAutoConfigurationTest.kt`
+
+## Related rules
+
+Also fetch:
+- `options` when changing `enableQueueTracing` or configuration surfaces.
+- `scopes` when changing consumer scope forking/lifecycle.
+- `opentelemetry` when changing coexistence with OTel auto-instrumentation.
+- `api` when changing public Kafka APIs or option methods.

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Features
 
+- Add Kafka queue tracing for Spring Boot 2 ([#5352](https://github.com/getsentry/sentry-java/pull/5352))
 - Add Kafka queue tracing for Spring Boot 4 ([#5348](https://github.com/getsentry/sentry-java/pull/5348))
 - Add `sentry-kafka` module for Kafka queue instrumentation without Spring ([#5288](https://github.com/getsentry/sentry-java/pull/5288))
 - Add Kafka queue tracing for Spring Boot 3 ([#5254](https://github.com/getsentry/sentry-java/pull/5254)), ([#5255](https://github.com/getsentry/sentry-java/pull/5255)), ([#5256](https://github.com/getsentry/sentry-java/pull/5256))

gradle/libs.versions.toml

@@ -183,6 +183,7 @@ springboot3-starter-security = { module = "org.springframework.boot:spring-boot-
 springboot3-starter-jdbc = { module = "org.springframework.boot:spring-boot-starter-jdbc", version.ref = "springboot3" }
 springboot3-starter-actuator = { module = "org.springframework.boot:spring-boot-starter-actuator", version.ref = "springboot3" }
 springboot3-starter-cache = { module = "org.springframework.boot:spring-boot-starter-cache", version.ref = "springboot3" }
+spring-kafka2 = { module = "org.springframework.kafka:spring-kafka", version = "2.8.11" }
 spring-kafka3 = { module = "org.springframework.kafka:spring-kafka", version = "3.3.5" }
 spring-kafka4 = { module = "org.springframework.kafka:spring-kafka" }
 kafka-clients = { module = "org.apache.kafka:kafka-clients", version = "3.8.1" }

sentry-samples/sentry-samples-spring-boot-opentelemetry-noagent/build.gradle.kts

@@ -55,6 +55,10 @@ dependencies {
   implementation(projects.sentryOpentelemetry.sentryOpentelemetryAgentlessSpring)
   implementation(projects.sentryAsyncProfiler)
 
+  // kafka
+  implementation(libs.spring.kafka2)
+  implementation(projects.sentryKafka)
+
   // database query tracing
   implementation(projects.sentryJdbc)
   runtimeOnly(libs.hsqldb)

sentry-samples/sentry-samples-spring-boot-opentelemetry-noagent/src/main/java/io/sentry/samples/spring/boot/queues/kafka/KafkaConsumer.java

@@ -0,0 +1,19 @@
+package io.sentry.samples.spring.boot.queues.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("kafka")
+public class KafkaConsumer {
+
+  private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
+
+  @KafkaListener(topics = "sentry-topic", groupId = "sentry-sample-group")
+  public void listen(String message) {
+    logger.info("Received message: {}", message);
+  }
+}

sentry-samples/sentry-samples-spring-boot-opentelemetry-noagent/src/main/java/io/sentry/samples/spring/boot/queues/kafka/KafkaController.java

@@ -0,0 +1,26 @@
+package io.sentry.samples.spring.boot.queues.kafka;
+
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@Profile("kafka")
+@RequestMapping("/kafka")
+public class KafkaController {
+
+  private final KafkaTemplate<String, String> kafkaTemplate;
+
+  public KafkaController(KafkaTemplate<String, String> kafkaTemplate) {
+    this.kafkaTemplate = kafkaTemplate;
+  }
+
+  @GetMapping("/produce")
+  String produce(@RequestParam(defaultValue = "hello from sentry!") String message) {
+    kafkaTemplate.send("sentry-topic", message);
+    return "Message sent: " + message;
+  }
+}

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

@@ -0,0 +1,12 @@
+# Kafka — activate with: --spring.profiles.active=kafka
+sentry.enable-queue-tracing=true
+
+spring.kafka.bootstrap-servers=localhost:9092
+spring.kafka.consumer.group-id=sentry-sample-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
+
+logging.level.org.apache.kafka=warn

sentry-samples/sentry-samples-spring-boot-opentelemetry-noagent/src/test/kotlin/io/sentry/systemtest/KafkaOtelCoexistenceSystemTest.kt

@@ -0,0 +1,41 @@
+package io.sentry.systemtest
+
+import io.sentry.systemtest.util.TestHelper
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import org.junit.Before
+
+class KafkaOtelCoexistenceSystemTest {
+  lateinit var testHelper: TestHelper
+
+  @Before
+  fun setup() {
+    testHelper = TestHelper("http://localhost:8080")
+    testHelper.reset()
+  }
+
+  @Test
+  fun `Sentry Kafka integration is suppressed when OTel is active`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("otel-coexistence-test")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.transaction == "GET /kafka/produce" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true &&
+        transaction.spans.any { span ->
+          span.op == "queue.publish" &&
+            span.origin == "auto.opentelemetry" &&
+            span.data?.get("messaging.system") == "kafka"
+        }
+    }
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.contexts.trace?.operation == "queue.process" &&
+        transaction.contexts.trace?.origin == "auto.opentelemetry" &&
+        transaction.contexts.trace?.data?.get("messaging.system") == "kafka" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true
+    }
+  }
+}

sentry-samples/sentry-samples-spring-boot-opentelemetry/build.gradle.kts

@@ -53,6 +53,10 @@ dependencies {
   implementation(projects.sentryAsyncProfiler)
   implementation(libs.otel)
 
+  // kafka
+  implementation(libs.spring.kafka2)
+  implementation(projects.sentryKafka)
+
   // database query tracing
   implementation(projects.sentryJdbc)
   runtimeOnly(libs.hsqldb)

sentry-samples/sentry-samples-spring-boot-opentelemetry/src/main/java/io/sentry/samples/spring/boot/queues/kafka/KafkaConsumer.java

@@ -0,0 +1,19 @@
+package io.sentry.samples.spring.boot.queues.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("kafka")
+public class KafkaConsumer {
+
+  private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
+
+  @KafkaListener(topics = "sentry-topic", groupId = "sentry-sample-group")
+  public void listen(String message) {
+    logger.info("Received message: {}", message);
+  }
+}

sentry-samples/sentry-samples-spring-boot-opentelemetry/src/main/java/io/sentry/samples/spring/boot/queues/kafka/KafkaController.java

@@ -0,0 +1,26 @@
+package io.sentry.samples.spring.boot.queues.kafka;
+
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@Profile("kafka")
+@RequestMapping("/kafka")
+public class KafkaController {
+
+  private final KafkaTemplate<String, String> kafkaTemplate;
+
+  public KafkaController(KafkaTemplate<String, String> kafkaTemplate) {
+    this.kafkaTemplate = kafkaTemplate;
+  }
+
+  @GetMapping("/produce")
+  String produce(@RequestParam(defaultValue = "hello from sentry!") String message) {
+    kafkaTemplate.send("sentry-topic", message);
+    return "Message sent: " + message;
+  }
+}

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

@@ -0,0 +1,12 @@
+# Kafka — activate with: --spring.profiles.active=kafka
+sentry.enable-queue-tracing=true
+
+spring.kafka.bootstrap-servers=localhost:9092
+spring.kafka.consumer.group-id=sentry-sample-group
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
+
+logging.level.org.apache.kafka=warn

sentry-samples/sentry-samples-spring-boot-opentelemetry/src/test/kotlin/io/sentry/systemtest/KafkaOtelCoexistenceSystemTest.kt

@@ -0,0 +1,41 @@
+package io.sentry.systemtest
+
+import io.sentry.systemtest.util.TestHelper
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import org.junit.Before
+
+class KafkaOtelCoexistenceSystemTest {
+  lateinit var testHelper: TestHelper
+
+  @Before
+  fun setup() {
+    testHelper = TestHelper("http://localhost:8080")
+    testHelper.reset()
+  }
+
+  @Test
+  fun `Sentry Kafka integration is suppressed when OTel is active`() {
+    val restClient = testHelper.restClient
+
+    restClient.produceKafkaMessage("otel-coexistence-test")
+    assertEquals(200, restClient.lastKnownStatusCode)
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.transaction == "GET /kafka/produce" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true &&
+        transaction.spans.any { span ->
+          span.op == "queue.publish" &&
+            span.origin == "auto.opentelemetry" &&
+            span.data?.get("messaging.system") == "kafka"
+        }
+    }
+
+    testHelper.ensureTransactionReceived { transaction, _ ->
+      transaction.contexts.trace?.operation == "queue.process" &&
+        transaction.contexts.trace?.origin == "auto.opentelemetry" &&
+        transaction.contexts.trace?.data?.get("messaging.system") == "kafka" &&
+        transaction.sdk?.integrationSet?.contains("SpringKafka") != true
+    }
+  }
+}

sentry-samples/sentry-samples-spring-boot/build.gradle.kts

@@ -43,6 +43,10 @@ dependencies {
   implementation(libs.springboot.starter.cache)
   implementation(libs.springboot.starter.websocket)
   implementation(libs.caffeine)
+
+  // kafka
+  implementation(libs.spring.kafka2)
+  implementation(projects.sentryKafka)
   implementation(Config.Libs.aspectj)
   implementation(Config.Libs.kotlinReflect)
   implementation(kotlin(Config.kotlinStdLib, KotlinCompilerVersion.VERSION))

sentry-samples/sentry-samples-spring-boot/src/main/java/io/sentry/samples/spring/boot/queues/kafka/KafkaConsumer.java

@@ -0,0 +1,19 @@
+package io.sentry.samples.spring.boot.queues.kafka;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.annotation.Profile;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+@Component
+@Profile("kafka")
+public class KafkaConsumer {
+
+  private static final Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);
+
+  @KafkaListener(topics = "sentry-topic", groupId = "sentry-sample-group")
+  public void listen(String message) {
+    logger.info("Received message: {}", message);
+  }
+}